Agent工程化演进实战：从提示词工程到上下文管理的架构升级

1次阅读

共计 2166 个字符，预计需要花费 6 分钟才能阅读完成。

在当前的 LLM 应用开发中，我们经常会遇到三个主要问题：

提示词版本管理混乱 ：随着业务需求的变化，提示词频繁调整却缺乏版本控制，导致模型输出效果不稳定，回溯困难。
长上下文效率低下 ：当处理长对话或复杂任务时，传统的滑动窗口方法会造成关键信息丢失，且全量上下文参与计算会导致 token 消耗激增。
状态维护耦合度高 ：Agent 的业务逻辑与对话状态管理紧密耦合，使得系统难以扩展和维护，特别是在多轮复杂交互场景下。

我们提出将 Agent 系统划分为三个独立层：

graph TD
    A[提示词工程层] -->| 模板化输入 | B[上下文工程层]
    B -->| 精炼上下文 | C[驾驭工程层]
    C -->| 决策输出 | A

提示词工程层 ：负责模板管理、版本控制和动态渲染
上下文工程层 ：处理上下文压缩、向量化检索和优先级排序
驾驭工程层 ：实现状态机管理、业务流程编排和异常处理

传统单体 Agent：
所有功能耦合在单一模块
上下文直接拼接字符串
状态管理散落在业务代码中
工程化方案：
职责边界清晰
支持组件独立升级
性能指标可单独监控

class PromptLoader:
    def __init__(self, max_size=100):
        self.cache = OrderedDict()
        self.max_size = max_size

    def get_prompt(self, version, params):
        cache_key = f"{version}:{json.dumps(params)}"
        if cache_key in self.cache:
            self.cache.move_to_end(cache_key)
            return self.cache[cache_key]

        # 模拟从数据库加载
        template = db.get_template(version)
        rendered = template.render(**params)

        if len(self.cache) >= self.max_size:
            self.cache.popitem(last=False)
        self.cache[cache_key] = rendered
        return rendered

时间复杂度分析：
– get 操作：O(1) 哈希查找
– 缓存淘汰：O(1) 双向链表操作

def compress_context(messages, embedding_model, top_k=3):
    # 提取关键对话轮次
    embeddings = [embedding_model.encode(msg) for msg in messages]
    similarity = cosine_similarity(embeddings)

    # 构建重要性图谱
    importance = np.sum(similarity, axis=1)
    important_indices = np.argsort(importance)[-top_k:]

    return [messages[i] for i in sorted(important_indices)]

算法效率：
– 嵌入计算：O(n) 取决于模型
– 相似度矩阵：O(n^2) 但 n 通常小于 100

class AgentStateMachine:
    def __init__(self):
        self.state = "INIT"
        self.handlers = {
            "INIT": self._handle_init,
            "PROCESSING": self._handle_processing,
            "CONFIRMATION": self._handle_confirmation
        }

    def transition(self, event):
        handler = self.handlers.get(self.state)
        if not handler:
            raise InvalidStateError()
        return handler(event)

    def _handle_init(self, event):
        if event["type"] == "user_input":
            self.state = "PROCESSING"
            return {"action": "process_request"}