引言：提示词版本控制的迫切性
各位同仁，大家好。今天我们将深入探讨一个在大型语言模型（LLM）应用开发中日益凸显的关键问题：提示词（Prompt）的版本控制。随着我们对LLM的依赖加深，提示词不再是简单的输入字符串，它们是精心设计的指令、上下文和示例，直接决定了模型行为和应用性能。尤其是在 LangGraph 这样复杂的协调框架中，一个应用可能包含数十个甚至上百个节点，每个节点都可能依赖于一个或多个提示词。管理这些提示词，就如同管理传统软件项目的代码库一样，面临着巨大的挑战。
想象一下以下场景：
- 迭代与实验的混乱： 你的团队正在尝试不同的提示词策略来优化某个 LangGraph 代理的决策逻辑。每个人都在本地修改提示词，然后部署测试。很快，你就不知道哪个版本在生产环境表现最好，哪个版本带来了回归。
- 团队协作的障碍： 多个开发者同时改进同一个代理的不同部分。一个开发者修改了一个核心提示词，另一个开发者在不知情的情况下基于旧版本进行了开发。当他们试图整合工作时，冲突和覆盖在所难免。
- 回溯与故障排查的困境： 生产环境出现问题，发现是某个提示词的微小改动导致模型行为异常。如果没有明确的版本历史，你很难快速定位问题根源，更难以回滚到稳定版本。
- A/B 测试与部署的挑战： 你想对两个不同版本的提示词进行A/B测试，或者将一个经过验证的提示词版本部署到生产环境。手动复制粘贴和命名约定极易出错，且难以自动化。
这些痛点共同指向一个核心需求：我们需要一套健壮的机制来管理提示词的生命周期。而当我们提到“版本控制”，自然会联想到软件开发领域的黄金标准——Git。Git 提供的分支、合并、提交、回滚等能力，正是我们管理提示词所需要的。
因此，今天的讲座，我们的目标是在 LangGraph 的语境下，构建一套“提示词版本控制”（Prompt Version Control, PVC）系统，使其具备类似 Git 的核心功能。这将不仅仅是一个理论探讨，更是一次深入代码的实践，我们将从数据模型、存储机制到核心操作，一步步构建这个系统，并演示它如何与 LangGraph 无缝集成。
核心概念与设计哲学
在深入代码之前，我们必须明确 PVC 系统的核心概念和设计哲学。我们的目标是借鉴 Git 的精髓，将其应用于提示词这一特定资产。
1. 提示词即数据/代码
我们将提示词视为第一类公民，它们是应用程序逻辑不可或缺的一部分，与源代码具有同等的重要性。这意味着提示词应该：
- 可被版本化： 每次有意义的更改都应被记录和追踪。
- 可被共享： 团队成员可以轻松地访问、修改和贡献提示词。
- 可被审查： 更改可以被审核，以确保质量和一致性。
- 可被部署： 特定版本的提示词可以被可靠地部署到不同的环境中。
2. Git 类比与核心功能
我们将 Git 的核心概念映射到 PVC 系统中：
| Git 概念 | PVC 对应概念 | 描述 |
|---|---|---|
| 仓库 (Repository) | PVC 仓库 | 包含所有提示词的版本历史、分支信息和配置。 |
| 对象 (Objects) | 提示词版本 (PromptVersion) | 存储特定时间点的一个提示词的完整内容（模板、变量、元数据）。具有不变性，通过内容哈希引用。 |
| 提交 (Commit) | 记录一次或多次提示词更改的快照。包含作者、时间戳、提交消息、父提交指针，以及一个指向当前提示词状态“树”的指针。 | |
| 引用 (References) | 分支 (Branch) | 一个指向特定提交的可变指针。允许在不影响主线开发的情况下进行并行实验。 |
| HEAD | 指向当前工作目录所基于的提交或分支。 | |
| 操作 (Operations) | 初始化 (Init) | 创建一个新的 PVC 仓库。 |
| 暂存 (Stage) | 将对提示词的修改标记为要包含在下一次提交中。 | |
| 提交 (Commit) | 将暂存的更改永久保存到仓库历史中。 | |
| 分支 (Branch) | 从当前提交创建一个新的开发线。 | |
| 切换 (Checkout) | 切换到不同的分支或特定的提交，更新工作目录中的提示词到该版本。 | |
| 日志 (Log) | 查看提交历史记录。 | |
| 差异 (Diff) | 比较两个提示词版本或两个提交之间的提示词变化。 | |
| 合并 (Merge) | 将一个分支的更改集成到另一个分支中。 | |
| 回滚 (Rollback) | 撤销一个或多个提交，恢复到之前的状态。 | |
| 标签 (Tag) | 为重要的提交（如发布版本）打上一个不可变的别名。 |
3. 不变性与内容寻址
如同 Git，我们的 PVC 系统将利用不变性原则。每一个提示词的版本和每一个提交对象都是不可变的。它们的内容一旦创建就不能修改。对内容的任何修改都会生成一个新的版本和新的哈希值。这种设计带来诸多好处：
- 完整性： 确保历史记录的真实性，任何篡改都会改变哈希值。
- 效率： 可以通过哈希值快速判断两个对象是否相同，避免重复存储。
- 简化： 避免了复杂的版本冲突管理，因为所有对象都是独立的。
我们将使用内容哈希（例如 SHA-256）作为每个对象的唯一标识符。
4. LangGraph 集成目标
PVC 系统的最终目的是为 LangGraph 提供稳定、可控的提示词源。这意味着：
- LangGraph 的节点应该能够通过 PVC 管理器获取到特定版本的提示词。
- 切换 LangGraph 代理所使用的提示词版本，应该像切换 Git 分支一样简单。
- PVC 系统应提供 Python 接口，方便在 LangGraph 应用中调用。
架构设计：PVC 系统的骨架
现在，我们来设计 PVC 系统的具体架构。我们将采用一个基于文件系统的存储后端，这足够简单直观，便于理解和实现，同时也能灵活扩展到数据库或其他存储方案。
1. 数据模型 (Pydantic)
我们首先定义 Pydantic 模型，它们将作为我们 PVC 系统中各种对象的结构化表示。
import hashlib
import json
import os
import shutil
import time
from datetime import datetime
from typing import Dict, List, Optional, Any
from pydantic import BaseModel, Field, PrivateAttr
# --- 辅助函数 ---
def _hash_object(data: Dict[str, Any]) -> str:
"""计算对象的SHA-256哈希值"""
# 确保字典键有序，以获得一致的哈希值
serialized_data = json.dumps(data, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(serialized_data.encode('utf-8')).hexdigest()
def _read_json_file(file_path: str) -> Dict[str, Any]:
"""读取JSON文件内容"""
if not os.path.exists(file_path):
return {}
with open(file_path, 'r', encoding='utf-8') as f:
return json.load(f)
def _write_json_file(file_path: str, data: Dict[str, Any]):
"""写入JSON文件内容"""
os.makedirs(os.path.dirname(file_path), exist_ok=True)
with open(file_path, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2, ensure_ascii=False)
# --- 核心数据模型 ---
class PromptContent(BaseModel):
"""
表示一个提示词的实际内容。
可以是简单的字符串，也可以是更复杂的模板结构。
"""
template: str = Field(..., description="提示词模板字符串，可包含占位符")
input_variables: List[str] = Field(default_factory=list, description="提示词模板中的输入变量列表")
metadata: Dict[str, Any] = Field(default_factory=dict, description="任意附加元数据")
def __str__(self):
return self.template
def format(self, **kwargs):
"""格式化提示词模板"""
return self.template.format(**kwargs)
class PromptVersion(BaseModel):
"""
一个具体的提示词版本，包含内容和其哈希值。
它是不可变的。
"""
content: PromptContent
hash: str = Field(..., description="该提示词内容的SHA-256哈希值")
created_at: float = Field(default_factory=time.time, description="创建时间戳")
@classmethod
def create(cls, content: PromptContent) -> 'PromptVersion':
"""创建PromptVersion对象并计算哈希值"""
content_dict = content.model_dump(exclude_defaults=True)
content_hash = _hash_object(content_dict)
return cls(content=content, hash=content_hash)
class TreeObject(BaseModel):
"""
表示一个目录树，映射提示词名称到其对应的 PromptVersion 哈希。
一个提交会指向一个TreeObject。
"""
prompts: Dict[str, str] = Field(default_factory=dict, description="映射提示词名称到PromptVersion哈希")
hash: str = Field(..., description="该TreeObject内容的SHA-256哈希值")
@classmethod
def create(cls, prompts: Dict[str, str]) -> 'TreeObject':
"""创建TreeObject对象并计算哈希值"""
tree_dict = {"prompts": prompts}
tree_hash = _hash_object(tree_dict)
return cls(prompts=prompts, hash=tree_hash)
class Commit(BaseModel):
"""
表示一次提交，包含元数据和指向TreeObject的哈希。
它是不可变的。
"""
hash: str = Field(..., description="该提交内容的SHA-256哈希值")
tree_hash: str = Field(..., description="指向该提交对应的TreeObject的哈希")
parent_hashes: List[str] = Field(default_factory=list, description="父提交的哈希列表（通常为1个，合并时可能多个）")
author: str = Field("unknown", description="提交者")
message: str = Field("initial commit", description="提交消息")
timestamp: float = Field(default_factory=time.time, description="提交时间戳")
@classmethod
def create(
cls,
tree_hash: str,
parent_hashes: Optional[List[str]] = None,
author: str = "PVC_System",
message: str = "No message",
) -> 'Commit':
"""创建Commit对象并计算哈希值"""
if parent_hashes is None:
parent_hashes = []
commit_data = {
"tree_hash": tree_hash,
"parent_hashes": sorted(parent_hashes), # 确保一致性
"author": author,
"message": message,
"timestamp": time.time(),
}
commit_hash = _hash_object(commit_data)
return cls(hash=commit_hash, **commit_data)
class Branch(BaseModel):
"""
表示一个分支，指向一个Commit哈希。
"""
name: str
commit_hash: str
class RepositoryConfig(BaseModel):
"""
仓库的全局配置。
"""
active_branch: str = "main"
head_commit: Optional[str] = None # 如果是detached HEAD
repo_path: str # 仓库根目录
说明：
PromptContent: 存储提示词的实际内容，包括模板字符串、期望的输入变量和任何自定义元数据。这使得提示词不仅限于简单字符串，可以支持更复杂的 LangChainPromptTemplate结构。PromptVersion: 包裹PromptContent，并计算其内容的 SHA-256 哈希值。这个哈希值是提示词版本的唯一标识符，确保了内容的不变性。TreeObject: 这是一个关键概念，类似 Git 的“树对象”。它存储了在某个特定时间点，仓库中所有（或部分）提示词的名称到其PromptVersion哈希的映射。一个提交会指向一个TreeObject。Commit: 表示一次提交。它包含了提交的元数据（作者、消息、时间戳），以及一个指向TreeObject的哈希值，该TreeObject定义了这次提交时所有提示词的状态。parent_hashes字段允许我们构建提交历史图。Branch: 一个简单的引用，存储分支名称和它当前指向的Commit哈希。RepositoryConfig: 存储仓库的全局配置，如当前活跃分支。
2. 存储层：文件系统结构
我们将 PVC 仓库的数据存储在一个特定的目录结构中，类似于 Git 的 .git 目录。
.pvc_repo/
├── config.json # 仓库配置 (RepositoryConfig)
├── HEAD # 指向当前活跃分支 (e.g., refs/heads/main) 或某个 commit hash
├── objects/ # 存储所有 PromptVersion, TreeObject, Commit 对象
│ ├── <hash_prefix>/ # 为了避免单个目录文件过多，按哈希前缀分目录
│ │ └── <full_hash>.json # 存储具体的 PromptVersion, TreeObject 或 Commit 对象内容
│ └── ...
└── refs/
└── heads/ # 存储分支引用
├── main # 存储 main 分支当前指向的 commit hash
└── feature-x # 存储 feature-x 分支当前指向的 commit hash说明：
.pvc_repo/: 仓库的根目录，所有 PVC 相关数据都存储在这里。config.json: 存储RepositoryConfig实例的 JSON 序列化数据。HEAD: 一个文本文件，内容是ref: refs/heads/<branch_name>（指向分支）或一个commit_hash（分离头指针状态）。objects/: 这是核心存储区域，所有PromptVersion、TreeObject和Commit对象都以其哈希值作为文件名存储在这里。我们使用哈希的前两位作为子目录，以避免单个目录文件过多。refs/heads/: 存储分支引用。每个文件以分支名称命名，内容是该分支当前指向的Commit哈希。
3. 核心管理器 (PVCManager)
PVCManager 类将是整个 PVC 系统的核心，负责协调所有操作。它将管理仓库的状态，并提供与外部交互的接口。
class PVCManager:
"""
Prompt Version Control (PVC) 系统的核心管理器。
提供 Git 类似的操作接口，用于管理提示词的版本。
"""
def __init__(self, repo_path: str):
self.repo_path = os.path.abspath(repo_path)
self.pvc_dir = os.path.join(self.repo_path, ".pvc_repo")
self.objects_dir = os.path.join(self.pvc_dir, "objects")
self.heads_dir = os.path.join(self.pvc_dir, "refs", "heads")
self.head_file = os.path.join(self.pvc_dir, "HEAD")
self.config_file = os.path.join(self.pvc_dir, "config.json")
self._config: Optional[RepositoryConfig] = None
self._current_tree_cache: Dict[str, PromptVersion] = {} # 缓存当前工作树的提示词
if os.path.exists(self.pvc_dir):
self._load_config()
self._load_current_tree() # 尝试加载当前HEAD指向的提示词
else:
print(f"Warning: PVC repository not initialized at {repo_path}. Call init_repo().")
def _load_config(self):
"""加载仓库配置"""
config_data = _read_json_file(self.config_file)
if config_data:
self._config = RepositoryConfig(**config_data)
self._config.repo_path = self.repo_path # 确保repo_path一致
else:
raise FileNotFoundError(f"Repository config file not found at {self.config_file}")
def _save_config(self):
"""保存仓库配置"""
if self._config:
_write_json_file(self.config_file, self._config.model_dump())
def _get_object_path(self, obj_hash: str) -> str:
"""根据哈希值获取对象存储路径"""
return os.path.join(self.objects_dir, obj_hash[:2], f"{obj_hash}.json")
def _write_object(self, obj: BaseModel) -> str:
"""将Pydantic对象写入对象存储，并返回其哈希值"""
obj_data = obj.model_dump(exclude_defaults=True)
obj_hash = obj_data.get('hash') # 假定对象本身已经计算了哈希
if not obj_hash:
raise ValueError("Object must have a 'hash' field to be written.")
path = self._get_object_path(obj_hash)
_write_json_file(path, obj_data)
return obj_hash
def _read_object(self, obj_hash: str, obj_type: type[BaseModel]) -> BaseModel:
"""从对象存储读取Pydantic对象"""
path = self._get_object_path(obj_hash)
if not os.path.exists(path):
raise FileNotFoundError(f"Object with hash {obj_hash} not found at {path}")
data = _read_json_file(path)
return obj_type(**data)
def _get_head_ref(self) -> str:
"""获取HEAD文件内容，即当前指向的引用或哈希"""
if not os.path.exists(self.head_file):
return ""
with open(self.head_file, 'r', encoding='utf-8') as f:
return f.read().strip()
def _update_head_ref(self, ref_content: str):
"""更新HEAD文件内容"""
_write_json_file(self.head_file, ref_content) # Actually write string, not JSON
with open(self.head_file, 'w', encoding='utf-8') as f:
f.write(ref_content)
def _get_branch_commit_hash(self, branch_name: str) -> Optional[str]:
"""获取指定分支的最新提交哈希"""
branch_path = os.path.join(self.heads_dir, branch_name)
if not os.path.exists(branch_path):
return None
with open(branch_path, 'r', encoding='utf-8') as f:
return f.read().strip()
def _update_branch_commit_hash(self, branch_name: str, commit_hash: str):
"""更新指定分支的最新提交哈希"""
os.makedirs(self.heads_dir, exist_ok=True)
branch_path = os.path.join(self.heads_dir, branch_name)
with open(branch_path, 'w', encoding='utf-8') as f:
f.write(commit_hash)
def _get_current_head_commit_hash(self) -> Optional[str]:
"""获取当前HEAD指向的实际提交哈希"""
head_ref = self._get_head_ref()
if head_ref.startswith("ref: "):
branch_name = head_ref[len("ref: refs/heads/"):]
return self._get_branch_commit_hash(branch_name)
elif head_ref:
# Detached HEAD
return head_ref
return None
def _get_commit(self, commit_hash: str) -> Commit:
"""根据哈希获取Commit对象"""
return self._read_object(commit_hash, Commit)
def _get_tree_object(self, tree_hash: str) -> TreeObject:
"""根据哈希获取TreeObject对象"""
return self._read_object(tree_hash, TreeObject)
def _get_prompt_version(self, prompt_version_hash: str) -> PromptVersion:
"""根据哈希获取PromptVersion对象"""
return self._read_object(prompt_version_hash, PromptVersion)
def init_repo(self, author: str = "PVC_System") -> str:
"""
初始化一个新的PVC仓库。
如果仓库已存在，则不做任何操作。
"""
if os.path.exists(self.pvc_dir):
print(f"Repository already initialized at {self.repo_path}")
self._load_config()
self._load_current_tree()
return self.repo_path
os.makedirs(self.objects_dir, exist_ok=True)
os.makedirs(self.heads_dir, exist_ok=True)
# 创建初始配置
self._config = RepositoryConfig(repo_path=self.repo_path, active_branch="main")
self._save_config()
# 创建空的初始 TreeObject
empty_tree = TreeObject.create(prompts={})
self._write_object(empty_tree)
# 创建初始 Commit
initial_commit = Commit.create(
tree_hash=empty_tree.hash,
author=author,
message="Initial commit"
)
self._write_object(initial_commit)
# 设置 main 分支指向初始 Commit
self._update_branch_commit_hash("main", initial_commit.hash)
# 设置 HEAD 指向 main 分支
self._update_head_ref("ref: refs/heads/main")
print(f"Initialized empty PVC repository at {self.repo_path}")
self._load_current_tree() # 加载初始树
return self.repo_path
def _load_current_tree(self):
"""
加载当前HEAD指向的提交中的所有提示词到缓存。
"""
self._current_tree_cache = {}
head_commit_hash = self._get_current_head_commit_hash()
if not head_commit_hash:
return
current_commit = self._get_commit(head_commit_hash)
current_tree = self._get_tree_object(current_commit.tree_hash)
for prompt_name, prompt_hash in current_tree.prompts.items():
self._current_tree_cache[prompt_name] = self._get_prompt_version(prompt_hash)
def get_current_prompts_tree(self) -> Dict[str, PromptVersion]:
"""
获取当前HEAD指向的提交中所有提示词的PromptVersion对象。
"""
return self._current_tree_cache
def get_prompt(self, prompt_name: str, branch: Optional[str] = None, commit_hash: Optional[str] = None) -> Optional[PromptContent]:
"""
获取指定名称的提示词。
优先使用 commit_hash，其次是 branch，最后是当前活跃分支。
"""
target_commit_hash = None
if commit_hash:
target_commit_hash = commit_hash
elif branch:
target_commit_hash = self._get_branch_commit_hash(branch)
if not target_commit_hash:
print(f"Error: Branch '{branch}' not found.")
return None
elif self._config and self._config.active_branch:
target_commit_hash = self._get_current_head_commit_hash()
else:
print("Error: No active branch or commit specified.")
return None
if not target_commit_hash:
return None
try:
target_commit = self._get_commit(target_commit_hash)
target_tree = self._get_tree_object(target_commit.tree_hash)
prompt_version_hash = target_tree.prompts.get(prompt_name)
if prompt_version_hash:
prompt_version = self._get_prompt_version(prompt_version_hash)
return prompt_version.content
else:
print(f"Prompt '{prompt_name}' not found in commit {target_commit_hash}")
return None
except FileNotFoundError as e:
print(f"Error getting prompt: {e}")
return None
def add_prompt_to_staging(self, prompt_name: str, content: PromptContent):
"""
将一个提示词添加到暂存区（in-memory），等待提交。
这会生成一个新的PromptVersion对象并缓存。
"""
prompt_version = PromptVersion.create(content)
self._current_tree_cache[prompt_name] = prompt_version
print(f"Staged prompt '{prompt_name}' (hash: {prompt_version.hash})")
def commit(self, message: str, author: str = "PVC_User") -> Optional[str]:
"""
将当前暂存区中的提示词状态创建一个新提交。
"""
if not self._config:
print("Error: Repository not initialized.")
return None
# 1. 保存所有新的 PromptVersion 对象
current_prompt_hashes = {}
for prompt_name, prompt_version in self._current_tree_cache.items():
self._write_object(prompt_version)
current_prompt_hashes[prompt_name] = prompt_version.hash
# 2. 创建并保存新的 TreeObject
new_tree = TreeObject.create(prompts=current_prompt_hashes)
self._write_object(new_tree)
# 3. 获取当前 HEAD 指向的父提交
parent_commit_hash = self._get_current_head_commit_hash()
parent_hashes = [parent_commit_hash] if parent_commit_hash else []
# 4. 创建并保存新的 Commit
new_commit = Commit.create(
tree_hash=new_tree.hash,
parent_hashes=parent_hashes,
author=author,
message=message
)
self._write_object(new_commit)
# 5. 更新当前分支的指针
head_ref = self._get_head_ref()
if head_ref.startswith("ref: refs/heads/"):
branch_name = head_ref[len("ref: refs/heads/"):]
self._update_branch_commit_hash(branch_name, new_commit.hash)
self._config.head_commit = None # Ensure it's not a detached HEAD
self._config.active_branch = branch_name
print(f"[{branch_name} {new_commit.hash[:7]}] {message}")
else:
# Detached HEAD state, update HEAD to point to new commit directly
self._update_head_ref(new_commit.hash)
self._config.head_commit = new_commit.hash
self._config.active_branch = "" # No active branch
print(f"[DETACHED HEAD {new_commit.hash[:7]}] {message}")
self._save_config()
self._load_current_tree() # 刷新缓存
return new_commit.hash
def branch(self, new_branch_name: str, start_point: Optional[str] = None) -> bool:
"""
创建新分支。
start_point 可以是 commit hash 或现有分支名。
如果未指定 start_point，则从当前HEAD创建。
"""
if not self._config:
print("Error: Repository not initialized.")
return False
if self._get_branch_commit_hash(new_branch_name):
print(f"Error: Branch '{new_branch_name}' already exists.")
return False
target_commit_hash = None
if start_point:
if self._get_branch_commit_hash(start_point): # start_point is a branch
target_commit_hash = self._get_branch_commit_hash(start_point)
else: # start_point is potentially a commit hash
try:
self._get_commit(start_point) # Validate commit exists
target_commit_hash = start_point
except FileNotFoundError:
print(f"Error: Start point '{start_point}' is neither a branch nor a valid commit hash.")
return False
else:
target_commit_hash = self._get_current_head_commit_hash()
if not target_commit_hash:
print("Error: Cannot create branch from an empty repository (no commits).")
return False
self._update_branch_commit_hash(new_branch_name, target_commit_hash)
print(f"Created new branch '{new_branch_name}' pointing to {target_commit_hash[:7]}")
return True
def checkout(self, target_ref: str) -> bool:
"""
切换到指定分支或提交。
target_ref 可以是分支名或 commit hash。
"""
if not self._config:
print("Error: Repository not initialized.")
return False
target_commit_hash = None
is_branch_checkout = False
# Try to resolve as a branch
branch_commit_hash = self._get_branch_commit_hash(target_ref)
if branch_commit_hash:
target_commit_hash = branch_commit_hash
is_branch_checkout = True
else:
# Try to resolve as a commit hash
try:
self._get_commit(target_ref) # Validate commit exists
target_commit_hash = target_ref
except FileNotFoundError:
print(f"Error: '{target_ref}' is neither a branch nor a valid commit hash.")
return False
if not target_commit_hash:
print(f"Error: Could not resolve target reference '{target_ref}'.")
return False
# Update HEAD
if is_branch_checkout:
self._update_head_ref(f"ref: refs/heads/{target_ref}")
self._config.active_branch = target_ref
self._config.head_commit = None
print(f"Switched to branch '{target_ref}'")
else:
self._update_head_ref(target_commit_hash)
self._config.active_branch = "" # Detached HEAD
self._config.head_commit = target_commit_hash
print(f"Checked out to commit '{target_ref[:7]}' (detached HEAD)")
self._save_config()
self._load_current_tree() # 刷新缓存
return True
def log(self, branch_name: Optional[str] = None, max_count: int = 10):
"""
打印提交历史。
如果指定 branch_name，则显示该分支的历史；否则显示当前HEAD的历史。
"""
start_commit_hash = None
if branch_name:
start_commit_hash = self._get_branch_commit_hash(branch_name)
if not start_commit_hash:
print(f"Error: Branch '{branch_name}' not found.")
return
else:
start_commit_hash = self._get_current_head_commit_hash()
if not start_commit_hash:
print("No commits found in current HEAD.")
return
current_commit_hash = start_commit_hash
count = 0
visited_commits = set() # To prevent infinite loops in case of cyclic graph (shouldn't happen with Git model)
print(f"Commit history for {branch_name if branch_name else 'HEAD'}:")
while current_commit_hash and count < max_count and current_commit_hash not in visited_commits:
try:
commit = self._get_commit(current_commit_hash)
timestamp_dt = datetime.fromtimestamp(commit.timestamp)
print(f"commit {commit.hash}")
print(f"Author: {commit.author}")
print(f"Date: {timestamp_dt.strftime('%Y-%m-%d %H:%M:%S')}")
print(f"n {commit.message}n")
visited_commits.add(current_commit_hash)
if commit.parent_hashes:
# For simplicity, just follow the first parent for linear log display
current_commit_hash = commit.parent_hashes[0]
else:
current_commit_hash = None
count += 1
except FileNotFoundError:
print(f"Error: Commit {current_commit_hash} not found in objects.")
break
except Exception as e:
print(f"Error processing commit {current_commit_hash}: {e}")
break
def diff_prompts(self, prompt_name: str, commit_a_hash: str, commit_b_hash: str):
"""
比较两个提交中指定提示词的差异。
"""
try:
commit_a = self._get_commit(commit_a_hash)
commit_b = self._get_commit(commit_b_hash)
tree_a = self._get_tree_object(commit_a.tree_hash)
tree_b = self._get_tree_object(commit_b.tree_hash)
prompt_hash_a = tree_a.prompts.get(prompt_name)
prompt_hash_b = tree_b.prompts.get(prompt_name)
content_a = self._get_prompt_version(prompt_hash_a).content.template if prompt_hash_a else f"Prompt '{prompt_name}' not found in {commit_a_hash[:7]}"
content_b = self._get_prompt_version(prompt_hash_b).content.template if prompt_hash_b else f"Prompt '{prompt_name}' not found in {commit_b_hash[:7]}"
print(f"Diff for prompt '{prompt_name}' between {commit_a_hash[:7]} and {commit_b_hash[:7]}:")
# Using a simple line-by-line diff for demonstration
lines_a = content_a.splitlines()
lines_b = content_b.splitlines()
max_len_a = max(len(line) for line in lines_a) if lines_a else 0
max_len_b = max(len(line) for line in lines_b) if lines_b else 0
from difflib import unified_diff
diff_lines = list(unified_diff(
lines_a, lines_b,
fromfile=f"a/{prompt_name} ({commit_a_hash[:7]})",
tofile=f"b/{prompt_name} ({commit_b_hash[:7]})",
lineterm=''
))
if not diff_lines:
print("No changes.")
else:
for line in diff_lines:
print(line)
except FileNotFoundError as e:
print(f"Error during diff: {e}")
except Exception as e:
print(f"An unexpected error occurred during diff: {e}")
def merge(self, source_branch: str, author: str = "PVC_User") -> Optional[str]:
"""
将源分支合并到当前活跃分支。
这是一个简化的合并实现，不处理复杂的文本冲突，默认“最新版本获胜”。
真实的 Git 合并涉及三方合并和冲突标记。
"""
if not self._config or not self._config.active_branch:
print("Error: No active branch to merge into. Checkout a branch first.")
return None
current_branch = self._config.active_branch
if current_branch == source_branch:
print("Error: Cannot merge a branch into itself.")
return None
source_commit_hash = self._get_branch_commit_hash(source_branch)
if not source_commit_hash:
print(f"Error: Source branch '{source_branch}' not found.")
return None
current_commit_hash = self._get_current_head_commit_hash()
if not current_commit_hash:
print("Error: Current branch has no commits.")
return None
# 获取两个分支的最新提示词树
source_commit = self._get_commit(source_commit_hash)
current_commit = self._get_commit(current_commit_hash)
source_tree = self._get_tree_object(source_commit.tree_hash)
current_tree = self._get_tree_object(current_commit.tree_hash)
# 简单合并策略：以 source_branch 的内容为准，如果当前分支也有，则覆盖。
# 更复杂的合并需要找到共同祖先，并进行三方合并。
merged_prompt_hashes = dict(current_tree.prompts) # Start with current branch's prompts
# Add/overwrite with source branch's prompts
for prompt_name, prompt_hash in source_tree.prompts.items():
merged_prompt_hashes[prompt_name] = prompt_hash
# In a real system, you'd detect conflicts here
if prompt_name in current_tree.prompts and current_tree.prompts[prompt_name] != prompt_hash:
print(f"Warning: Prompt '{prompt_name}' modified in both branches. Using '{source_branch}' version.")
# If the trees are identical, it's a fast-forward merge (or already merged)
if merged_prompt_hashes == current_tree.prompts:
print(f"Already up-to-date or fast-forward merge possible. No new commit created.")
if current_tree.hash != source_tree.hash: # Fast-forward merge
self._update_branch_commit_hash(current_branch, source_commit_hash)
self._update_head_ref(f"ref: refs/heads/{current_branch}")
self._config.head_commit = None
self._config.active_branch = current_branch
self._save_config()
self._load_current_tree()
print(f"Fast-forwarded branch '{current_branch}' to {source_commit_hash[:7]}.")
return source_commit_hash # Return hash of source, indicating fast-forward or no-op
# 2. 创建并保存新的 TreeObject
new_tree = TreeObject.create(prompts=merged_prompt_hashes)
self._write_object(new_tree)
# 3. 创建合并提交 (包含两个父提交)
merge_commit = Commit.create(
tree_hash=new_tree.hash,
parent_hashes=[current_commit_hash, source_commit_hash],
author=author,
message=f"Merge branch '{source_branch}' into '{current_branch}'"
)
self._write_object(merge_commit)
# 4. 更新当前分支的指针
self._update_branch_commit_hash(current_branch, merge_commit.hash)
self._update_head_ref(f"ref: refs/heads/{current_branch}")
self._config.head_commit = None
self._config.active_branch = current_branch
self._save_config()
self._load_current_tree() # 刷新缓存
print(f"Merged branch '{source_branch}' into '{current_branch}'. New commit: {merge_commit.hash[:7]}")
return merge_commit.hash
def rollback(self, target_commit_hash: str) -> bool:
"""
回滚当前分支到指定的提交。
这将创建一个新的提交，其状态与 target_commit_hash 相同。
这类似于 `git revert`，而不是 `git reset --hard`。
"""
if not self._config or not self._config.active_branch:
print("Error: No active branch to rollback. Checkout a branch first.")
return False
current_branch = self._config.active_branch
current_head_commit_hash = self._get_current_head_commit_hash()
if current_head_commit_hash == target_commit_hash:
print(f"Branch '{current_branch}' is already at commit {target_commit_hash[:7]}. No rollback needed.")
return True
try:
target_commit = self._get_commit(target_commit_hash)
target_tree = self._get_tree_object(target_commit.tree_hash)
# Stage the prompts from the target commit
self._current_tree_cache = {}
for prompt_name, prompt_hash in target_tree.prompts.items():
self._current_tree_cache[prompt_name] = self._get_prompt_version(prompt_hash)
# Create a new commit with the reverted state
rollback_commit_hash = self.commit(
message=f"Revert to commit {target_commit_hash[:7]}",
author=f"{self._config.active_branch} Rollback"
)
print(f"Rolled back branch '{current_branch}' to state of commit {target_commit_hash[:7]}. New rollback commit: {rollback_commit_hash[:7]}")
return True
except FileNotFoundError as e:
print(f"Error during rollback: {e}")
return False
except Exception as e:
print(f"An unexpected error occurred during rollback: {e}")
return False
def list_branches(self) -> List[str]:
"""列出所有分支"""
if not os.path.exists(self.heads_dir):
return []
return [f for f in os.listdir(self.heads_dir) if os.path.isfile(os.path.join(self.heads_dir, f))]
def get_active_branch(self) -> str:
"""获取当前活跃分支名称"""
if self._config and self._config.active_branch:
return self._config.active_branch
head_ref = self._get_head_ref()
if head_ref.startswith("ref: refs/heads/"):
return head_ref[len("ref: refs/heads/"):]
return "DETACHED_HEAD" # Indicate detached HEAD statePVCManager 的关键方法解释：
_hash_object,_read_json_file,_write_json_file: 辅助函数，用于对象的哈希计算和 JSON 文件的读写。_get_object_path,_write_object,_read_object: 负责对象存储的低层接口，将 Pydantic 对象序列化为 JSON 并存储到objects/目录，或反之。_get_head_ref,_update_head_ref,_get_branch_commit_hash,_update_branch_commit_hash: 管理HEAD文件和refs/heads/目录中的引用。_get_current_head_commit_hash,_get_commit,_get_tree_object,_get_prompt_version: 获取仓库中不同类型的对象。init_repo: 初始化一个新仓库，创建必要的目录结构和初始提交。_load_current_tree: 从当前 HEAD 指向的提交中加载所有提示词到内部缓存 (_current_tree_cache)，模拟 Git 的工作区。get_current_prompts_tree: 返回当前缓存的提示词树。get_prompt: 从仓库中获取指定名称的提示词内容，可以指定分支或提交哈希。add_prompt_to_staging: 将修改后的提示词添加到暂存区（实际上是更新_current_tree_cache）。commit: 核心操作。它会根据_current_tree_cache创建新的TreeObject和Commit，并将它们写入对象存储，然后更新当前分支的指针。branch: 创建一个新分支，使其指向当前 HEAD 或指定提交。checkout: 切换当前工作区到指定分支或提交。这会更新HEAD指针和_current_tree_cache。log: 遍历并打印提交历史。diff_prompts: 比较两个提交中某个提示词的模板差异。这里使用了difflib.unified_diff进行简单的行级别比较。merge: 将源分支合并到当前活跃分支。这是一个简化版本，它不处理复杂的文本冲突，而是简单地以源分支的提示词版本覆盖目标分支中同名的提示词。真正的 Git 合并是一个复杂的三方合并过程，涉及共同祖先的查找和冲突标记。rollback: 回滚到指定的提交。这类似于git revert，它不是直接修改历史，而是创建一个新的提交，其内容与目标提交相同。
LangGraph 集成实践
现在我们已经有了一个功能相对完善的 PVC 系统。下一步是将其与 LangGraph 结合起来。 LangGraph 的节点通常会在其内部逻辑中使用提示词。通过 PVCManager，我们可以让这些节点动态地获取和使用版本化的提示词。
我们将创建一个简单的 LangGraph 代理作为示例，其中一个节点需要一个由 PVC 管理的提示词。
首先，确保安装 LangChain 和 LangGraph：
pip install langchain langgraph pydantic
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
# 假设您已设置 OPENAI_API_KEY 环境变量
# --- 1. 定义 LangGraph 状态 ---
class AgentState(BaseModel):
messages: List[Any] = Field(default_factory=list)
tool_output: Optional[str] = None
# --- 2. 定义 LangGraph 节点函数 ---
# 这是一个使用 PVC 管理的提示词的节点
def call_llm_node(state: AgentState, pvc_manager: PVCManager):
"""
LLM 节点，使用 PVC 管理的提示词。
"""
# 从 PVC 获取提示词
llm_prompt_content = pvc_manager.get_prompt("llm_decision_prompt")
if not llm_prompt_content:
raise ValueError("LLM decision prompt not found in PVC!")
# 将 PromptContent 转换为 LangChain 的 ChatPromptTemplate
llm_prompt_template = ChatPromptTemplate.from_template(llm_prompt_content.template)
llm = ChatOpenAI(temperature=0)
# 构造消息历史，包括系统提示词
system_message = llm_prompt_template.format(
# 假设提示词模板包含一个 {context} 变量
context="You are a helpful assistant. Make a decision based on the user's request."
)
# Combine system message with actual user messages
full_messages = [("system", system_message)] + [
(msg.type, msg.content) for msg in state.messages if isinstance(msg, HumanMessage)
]
# 调用 LLM
response = llm.invoke(full_messages)
# 更新状态
return AgentState(messages=state.messages + [response])
def tool_node(state: AgentState):
"""
工具节点，模拟一个外部工具的调用。
"""
# 假设我们有一个工具，可以处理用户的请求
last_message = state.messages[-1].content
if "天气" in last_message:
tool_result = "今天天气晴朗，气温25度。"
else:
tool_result = f"执行了工具，处理了请求：'{last_message}'"
return AgentState(messages=state.messages + [HumanMessage(content=f"Tool Output: {tool_result}")], tool_output=tool_result)
# --- 3. 定义 LangGraph 路由逻辑 ---
def decide_next_step(state: AgentState):
last_message = state.messages[-1].content
if "工具" in last_message or "天气" in last_message:
return "tool"
return "llm" # 默认回到LLM进行更多思考或回复
# --- 4. 构建 LangGraph ---
def build_langgraph_agent(pvc_manager: PVCManager):
workflow = StateGraph(AgentState)
workflow.add_node("llm", lambda state: call_llm_node(state, pvc_manager))
workflow.add_node("tool", tool_node)
workflow.set_entry_point("llm")
workflow.add_conditional_edges(
"llm",
decide_next_step,
{"tool": "tool", "llm": "llm", END: END} # 假设END是LLM最终决定终止
)
workflow.add_edge("tool", "llm") # 工具执行完通常会回到LLM进行总结或下一步判断
app = workflow.compile()
return app
# --- 5. 演示 PVC 与 LangGraph 的集成 ---
if __name__ == "__main__":
repo_path = "./my_prompt_repo"
pvc = PVCManager(repo_path)
pvc.init_repo(author="LangGraph Developer")
# --- 阶段 1: 初始提示词版本 ---
print("n--- 阶段 1: 初始提示词版本 ---")
initial_prompt_content = PromptContent(
template="你是一个友好的AI助手。你的任务是根据用户请求进行回复，并决定是否需要使用工具。如果用户提到'工具'或'天气'，请指示使用工具。n用户请求: {input}n{context}",
input_variables=["input", "context"],
metadata={"version": "1.0", "purpose": "initial_response"}
)
pvc.add_prompt_to_staging("llm_decision_prompt", initial_prompt_content)
initial_commit_hash = pvc.commit("Add initial LLM decision prompt")
print(f"当前活跃分支: {pvc.get_active_branch()}")
print(f"初始提交哈希: {initial_commit_hash}")
# 构建并运行代理
agent_app_v1 = build_langgraph_agent(pvc)
print("n运行代理 (初始版本):")
for s in agent_app_v1.stream({"messages": [HumanMessage(content="你好，帮我查一下天气")]}):
print(s)
print("--- 代理 V1 运行结束 ---")
# --- 阶段 2: 创建新分支并改进提示词 ---
print("n--- 阶段 2: 创建新分支并改进提示词 ---")
pvc.branch("feature/improved_prompt")
pvc.checkout("feature/improved_prompt")
improved_prompt_content = PromptContent(
template="你是一个高级AI助手，专注于提供简洁且准确的回复。在需要时，你可以利用外部工具。如果用户的请求明确需要外部信息（如'天气'），请优先考虑工具。否则，请直接回复。n用户请求: {input}n{context}",
input_variables=["input", "context"],
metadata={"version": "1.1", "purpose": "more_concise_response"}
)
pvc.add_prompt_to_staging("llm_decision_prompt", improved_prompt_content)
improved_commit_hash = pvc.commit("Improve LLM decision prompt for conciseness")
print(f"当前活跃分支: {pvc.get_active_branch()}")
print(f"改进后提交哈希: {improved_commit_hash}")
# 构建并运行代理 (使用改进后的提示词)
agent_app_v2 = build_langgraph_agent(pvc)
print("n运行代理 (改进版本 - feature/improved_prompt 分支):")
for s in agent_app_v2.stream({"messages": [HumanMessage(content="我需要知道天气怎么样")]}):
print(s)
print("--- 代理 V2 运行结束 ---")
# --- 阶段 3: 切换回主分支并进行合并 ---
print("n--- 阶段 3: 切换回主分支并进行合并 ---")
pvc.checkout("main")
print(f"当前活跃分支: {pvc.get_active_branch()}")
# 查看差异
print("n查看 main 分支和 feature/improved_prompt 分支的提示词差异:")
pvc.diff_prompts("llm_decision_prompt", initial_commit_hash, improved_commit_hash)
# 合并
merge_commit_hash = pvc.merge("feature/improved_prompt")
print(f"合并后提交哈希: {merge_commit_hash}")
# 再次运行代理 (合并后)
agent_app_merged = build_langgraph_agent(pvc)
print("n运行代理 (合并后 - main 分支):")
for s in agent_app_merged.stream({"messages": [HumanMessage(content="今天的气候如何？")]}):
print(s)
print("--- 代理 V_Merged 运行结束 ---")
# --- 阶段 4: 回滚演示 ---
print("n--- 阶段 4: 回滚演示 ---")
# 假设我们发现合并后的提示词有问题，想回滚到初始状态
pvc.rollback(initial_commit_hash)
# 再次运行代理 (回滚后)
agent_app_rolled_back = build_langgraph_agent(pvc)
print("n运行代理 (回滚后 - main 分支):")
for s in agent_app_rolled_back.stream({"messages": [HumanMessage(content="帮我查天气")]}):
print(s)
print("--- 代理 V_RolledBack 运行结束 ---")
# --- 阶段 5: 查看历史 ---
print("n--- 阶段 5: 查看历史 ---")
pvc.log("main", max_count=5)
# 清理仓库 (可选)
# shutil.rmtree(repo_path)
# print(f"nCleaned up repository at {repo_path}")演示流程说明：
- 初始化仓库并添加初始提示词： 我们首先初始化一个 PVC 仓库，然后定义一个名为
llm_decision_prompt的初始提示词，并提交到main分支。 - 构建并运行初始代理： 使用
pvc管理器构建一个 LangGraph 代理，其llm节点会从pvc获取llm_decision_prompt。我们运行它来观察其行为。 - 创建分支并改进提示词： 我们创建一个名为
feature/improved_prompt的新分支，并切换到该分支。在这个分支上，我们修改llm_decision_prompt的内容，使其更简洁，然后提交这些更改。 - 运行改进版代理： 再次构建代理，由于我们当前在
feature/improved_prompt分支，代理会自动加载该分支上的最新提示词。 - 切换回主分支并合并： 我们切换回
main分支，然后使用diff_prompts查看main和feature/improved_prompt两个分支上llm_decision_prompt的差异。接着，我们将feature/improved_prompt分支合并到main。 - 运行合并后代理： 再次运行代理，确认
main分支现在包含了改进后的提示词。 - 回滚演示： 假设合并后发现问题，我们演示如何使用
rollback将main分支的状态回滚到最初的提交。 - 查看历史： 最后，使用
log命令查看main分支的提交历史，可以看到所有的操作都被清晰地记录下来。
通过这个例子，我们可以看到 PVCManager 如何作为一个中央枢纽，为 LangGraph 应用提供版本化的提示词。 LangGraph 节点不需要关心提示词的具体存储位置或版本切换逻辑，它们只需要通过 PVCManager.get_prompt() 方法获取所需的提示词即可。
高级考量与未来展望
我们已经构建了一个功能性 PVC 系统，但真实的生产环境往往需要更强大的功能和更细致的考虑。
1. 冲突解决策略
我们当前的 merge 实现非常简化，采用“源分支获胜”的策略。在实际开发中，当两个分支对同一个提示词进行了不同的修改时，需要：
- 检测冲突： 识别出哪些提示词被两个分支独立修改。
- 三方合并： 找到共同祖先，并尝试自动合并。
- 手动解决： 如果无法自动合并，则需要用户手动编辑提示词，解决冲突并重新提交。这可能需要一个专门的 CLI 或 UI 工具来指导用户。
- 合并驱动： 对于特定类型的提示词（例如，JSON 格式的少样本示例），可以编写自定义的合并驱动程序。
2. 存储后端多样性
当前系统使用文件系统作为存储后端。对于大型团队或高可用性需求，可以考虑：
- 关系型数据库 (SQLite, PostgreSQL): 更适合结构化查询、事务和索引。
- NoSQL 数据库 (MongoDB, DynamoDB): 适用于灵活的提示词结构和大规模存储。
- 云对象存储 (S3, GCS): 成本低廉，可扩展性强，适合存储大量提示词版本。
- 专用提示词管理平台集成: 如 Weights & Biases Prompts, PromptLayer 等，它们提供额外的可视化、评估和部署功能。
3. CLI/UI 工具
一个强大的 PVC 系统需要用户友好的交互界面。可以开发：
- 命令行工具 (CLI): 封装
PVCManager的方法，提供pvc init,pvc commit,pvc branch,pvc checkout,pvc diff,pvc merge,pvc log等命令。 - 图形用户界面 (GUI): 提供更直观的方式来查看提示词历史、分支图，以及可视化差异和解决冲突。
4. CI/CD 集成
将 PVC 集成到持续集成/持续部署 (CI/CD) 流程中是自动化 LLM 应用开发的关键：
- 自动化测试： 在每次提示词提交或合并后，自动运行测试（如单元测试、集成测试、评估指标测试）来验证提示词的质量和性能。
- 自动化部署： 只有通过所有测试的特定提示词版本才允许部署到预生产或生产环境。可以基于分支或标签进行部署。
- 版本回滚： 生产环境出现问题时，可以快速回滚到上一个稳定版本的提示词。
5. 提示词测试与评估
仅仅版本化提示词是不够的，我们还需要验证它们的有效性：
- 单元测试： 针对提示词的特定部分（如解析 JSON 输出）编写测试。
- 集成测试： 在 LangGraph 流程中测试整个代理与特定提示词的交互。
- 评估框架： 使用 RAGAS、LangChain Evaluator 等工具，针对一组评估数据集，量化不同提示词版本的性能指标（如准确性、相关性、流畅性）。这些评估结果可以作为提交元数据的一部分。
6. 多租户与权限管理
在企业环境中，可能需要支持多个项目或团队，并对提示词的访问和修改进行权限控制。这需要一个更复杂的存储和认证授权层。
7. 语义化版本控制与标签
除了分支，还可以为重要的提示词版本打上标签（如 v1.0.0, v1.1.0-beta），这对于发布管理和追溯尤为有用。
8. 提示词的结构化定义
目前我们使用 PromptContent 来存储提示词模板和输入变量。可以进一步规范化提示词的结构，例如使用 YAML/JSON Schema 定义提示词的输入、输出格式，以及少样本示例等，从而实现更强大的类型检查和自动化工具支持。
结束语
通过今天的探讨与实践，我们看到了如何借鉴 Git 的版本控制思想，为 LangGraph 应用中的提示词构建一套强大的管理系统。这不仅仅是技术上的挑战，更是 LLM 时代软件工程最佳实践的演进。一个健壮的提示词版本控制体系，将极大地提升 LLM 应用开发的效率、可靠性和团队协作能力，为构建可信赖、高性能的 AI 代理奠定坚实基础。