链载Ai

标题: AI Agent [打印本页]

---

作者: 链载Ai    时间: 昨天 21:32
标题: AI Agent

写在前面

👀

在Trajectory的实现中，参照了一些开源的实现，主要是字节的trae Agent和langgraph的源码。这里在看字节的Trae agent的过程中，也顺带改了一个小bug，下一篇会讲下trae agent的源码解析，主要聚焦在几个方面：

• prompt是怎么写的？
• benchmark是如何做的？
• 有一个奇怪的多轮对话的报错，是为什么？

另外下一步会看下PromptX的实现，据说是提出一个挑选工具的协议，看看PromptX这个项目是怎么实现的。

感兴趣的话，先关注吧，说实话，网上也很少有博主像我写的这么详细的。

背景

📌

Trajectory 是什么？它的核心价值是什么？

在 Agent 的语境下，Trajectory（轨迹）指的是 Agent 在处理一个任务或一次对话交互过程中的完整执行记录（轨迹）指的是 Agent 在处理一个任务或一次对话交互过程中的完整执行记录。它就像飞机的“黑匣子”，详细记录了从接收用户输入开始，到最终给出响应的每一步思考、决策和行动。

一个典型的 Trajectory 包含以下关键信息：

• 用户输入 (User Message)：对话的起点。
• AI 的思考过程 (Thought)：Agent 的中间推理步骤。
• 工具调用 (Tool Call)：Agent 决定使用哪个工具，以及传入的参数。
• 工具执行结果 (Tool Output)：工具返回给 Agent 的信息。
• AI 的最终响应 (AI Message)：Agent 回复给用户的内容。
  
  

其核心价值主要体现在以下几个方面：

• 调试与可观测性 (Debugging & Observability)：当 Agent 行为不符合预期时，Trajectory 是定位问题的最直接、最有效的工具。开发者可以清晰地看到每一步的输入输出，快速诊断是模型幻觉、工具错误还是逻辑流问题。
• 审计与归档 (Audit & Archive)：在金融、法务、客服等需要合规和追溯的场景下，Trajectory 提供了一份不可篡改的、详细的交互历史。这既可以作为审计凭证，也可以作为历史案例进行归档，计与归档 (Audit &Archive)：在金融、法务、客服等需要合规和追溯的场景下，Trajectory 提供了一份不可篡改的、详细的交互历史。这既可以作为审计凭证，也可以作为历史案例进行归档，用于后续的分析和复盘。
• 评估与优化 (Evaluation & Optimization)：通过分析大量的 Trajectory 数据，我们可以评估 Agent 在不同任务上的表现，发现其能力的边界和常见的失败模式，为后续的模型微调（Fine-tuning）或 Prompt Engineering 提供数据支持。

因此想基于 LangGraph 的 Agent 设计并实现一个健壮、可扩展的 Trajectory（轨迹）记录系统。该系统不仅能满足基本的调试、审计和归-档需求，还引入了分布式追踪理念，为 Agent 的行为提供了深度可观测性，最终实现了类似 LangSmith 的轨迹分组与追踪效果。

实现目标

核心目标是捕获 Agent在执行任务过程中的每一步关键信息，并将其结构化地记录下来。希望做到的功能：

• 调试与审计：开发者可以清晰地回溯 Agent 的思考链、工具调用和模型响应，快速定位问题。
• 归档与分析：将 Agent 的完整交互历史永久化存储，用于后续的行为分析和模型优化。
• 可观测性与分布式追踪：借鉴 OpenTelemetry 等分布式追踪系统的理念，为每一次用户交互（Trace）和其中的每一个步骤（Span）分配唯一 ID，实现跨组件、跨服务的行为链路追踪。
• 多轮对话分组：能准确地将一次完整的端到端对话（从用户输入到最终回复）划分为一个独立的 Trace，便于分组查看和分析。
• 高可扩展性：系统设计应与具体存储后端解耦，支持从本地文件轻松扩展到 Kafka、数据库或专业日志系统。
• 灵活集成：既能无缝集成到 LangGraph 的ReAct Agent 中，也能作为一个独立的节点（Node）在任何 LangGraph 图中即插即用。
  
  

设计与实现

设计

在实现 Trajectory 记录功能的过程中，我评估了多种技术方案，最终选择了一种基于“扫描 state['messages']”的模式Trajectory 记录功能的过程中，我评估了多种技术方案，最终选择了一种基于“扫描 state['messages']”的模式。这个决策是权衡了多种方案的利弊后做出的。

曾经考虑过的方案：

1. 节点装饰器 (Node Decorator)：最初，想为 LangGraph 中的每一个 Node（节点）都包裹一个装饰器。这个装饰器会在节点执行前后自动记录日志。

• 代码类似
• 优点：逻辑和业务分离，看起来很优雅。
• 缺点：实现起来非常复杂。LangGraph 的节点功能各异，有的调用 LLM，有的执行工具，有的只是简单的逻辑判断。为这些异构的节点设计一个通用的、能提取所有关键信息的装饰器，成本很高，且容易与 LangGraph 的内部机制产生冲突。

  class TrajectoryHook:  """Hook for automatically recording LangGraph execution."""
    def __init__(self, recorder: TrajectoryRecorder):    self.recorder = recorder    self._session_id: Optional[str] = None
    def wrap_node(self, node_name: str, node_func: Callable) -> Callable:    """Wrap a node function to record its execution."""    @wraps(node_func)    async def wrapped_node(state: Dict[str, Any]) -> Any:      if not self._session_id:        return await node_func(state)
        # Record node start      await self.recorder.record_event(        self._session_id,        node_name=node_name,        event_type="node_start",        data={"state_keys": list(state.keys()) if isinstance(state, dict) else None}      )
        try:        # Execute node        if asyncio.iscoroutinefunction(node_func):          result = await node_func(state)        else:          result = node_func(state)
          # Record node end        await self.recorder.record_event(          self._session_id,          node_name=node_name,          event_type="node_end",          data={"has_result": result is not None}        )
          # Record messages if present        if isinstance(result, dict) and "messages" in result:          messages = result["messages"]          if isinstance(messages, list):            for msg in messages:              if hasattr(msg, "content"): # 确保是消息对象                await self.recorder.record_message(self._session_id, msg)
          # Record node output        await self.recorder.record_node_output(          self._session_id,          node_name,          result        )
          return result
        except Exception as e:        # Record error        await self.recorder.record_error(          self._session_id,          error_type=type(e).__name__,          error_message=str(e),          node_name=node_name        )        raise
      return wrapped_node
    async def __aenter__(self):    """Start recording session."""    self._session_id = await self.recorder.start_session()    return self
    async def __aexit__(self, exc_type, exc_val, exc_tb):    """End recording session."""    if self._session_id:      success = exc_type is None      await self.recorder.end_session(self._session_id, success=success)      self._session_id = None

  
  
• 重写检查点 (RecordingCheckpoint)：另一个思路是包装 LangGraph 的 Checkpoint（检查点）机制。因为 Checkpoint 本身就是用来持久化 Agent 状态的，我们可以在它保存状态的同时，将轨迹信息也一并记录下来。
• 优点：这是一个非常可行的方案。Checkpoint 是状态变更的关键枢纽，在这里做文章能确保捕捉到所有重要的变化。
• 缺点：它将轨迹记录与状态持久化逻辑强耦合。如果我们未来想更换 Checkpoint 的后端（比如从内存换到 Redis），或者在某些场景下禁用 Checkpoint，可能会影响到轨迹记录功能。当然也可以做一个组合的checkpoint，也是一个可行的方案！
  
  
最终选择的方案：扫描state['messages']
我最终实现的方案，其核心依据是：
👀
无论 Agent 内部的逻辑图（Graph）多么复杂，其所有关键的外部交互（用户输入、AI回复、工具调用与结果）最终都会以消息（Message）的形式被追加到 state['messages'] 列表中。


基于这个洞察，我的实现变得非常纯粹和解耦：
• 我们设计了一个MessageProcessor，它的唯一职责就是对比上一次的状态和当前状态，找出messages列表中的**新消息**。
• 一旦发现新消息，就将其标准化为一条或多条轨迹事件（Event），并赋予它们正确的 Trace ID 和 Span ID。
• 这种方式不关心这些消息是哪个 Node 产生的，也不关心 Checkpoint 是否启用。它只关心最终的结果，从而实现了与 LangGraph 内部执行逻辑的解耦。

实现

为实现上述目标，系统被设计为由多个松散耦合的组件构成，每个组件承担独立的职责。
关键组件
1. TraceContext: 追踪上下文，一个轻量级的数据容器，负责在整个 LangGraph 的执行流程中传递追踪状态，主要包含 trace_id、span_id和 parent_sntext: 追踪上下文，一个轻量级的数据容器，负责在整个 LangGraph 的执行流程中传递追踪状态，主要包含 trace_id、span_id 和 parent_span_id。
2. TrajectoryRecorder: 轨迹记录器，是系统的核心协调者。它本身是无状态的，接收处理过的事件数据，并通过一个可插拔的 TrajectoryBackend 将数据写入到指定的存ecorder: 轨迹记录器，是系统的核心协调者。它本身是无状态的，接收处理过的事件数据，并通过一个可插拔的 TrajectoryBackend 将数据写入到指定的存储中。
3. TrajectoryBackend: 存储后端，定义了数据写入的接口。LocalFileBackendend: 存储后端，定义了数据写入的接口。LocalFileBackend 是其默认实现，将轨迹数据以 JSONL 格式写入本地文件，每一行代表一个事件。这种设计使得更换后端（如 Kafka、Redis、PostgreSQL）变得非常简单。
4. MessageProcessor: 消息处理器，负责将 LangGraphstate['messages']中的原始消息转换为结构化的、可记录的 Trajectory 事件。它能识别消息类型（human,ai, tool）并生成相应的事件 payload。
5. ReactTrajectoryHook**TrajectoryNode**: 集成层，负责将轨迹记录功能接入 LangGraph。
• ReactTrajectoryHook作为 ReAct Agent 的post_model_hook，在每次模型调用后触发，自动管理 Trace 的生命周期（开始与结束）。
• TrajectoryNode作为一个独立的 LangGraph 节点，可以在图的任意位置被调用，手动记录轨迹。
• TrajectoryViewer: 一个简单的命令行工具，用于读取本地存储的轨迹文件，并按照trace_id进行分组和格式化展示，模拟了 LangSmith 的可视化效果。
  
  

分布式追踪机制

为了实现类似 LangSmith 那样的可视化效果，我们需要将一次完整的“请求-响应”流程中的所有事件（多次工具调用、AI 思考等）聚合到同一个 Trace 下。这就引出了一个关键问题：如何判断一轮对话的开始和结束？
• 当前实现：我们的策略相对直接——**通过新的用户输入来开启一个新的 Trace**。当一个HumanMessage（用户消息）在上一轮对话结束后出现时，系统会生成一个新的trace_id。本轮中所有的后续事件（AIMessage、ToolCall、ToolMessage）都会沿用这个trace_id，但会拥有各自独立的span_id，并通过parent_span_id建立父子关系。
通过这三个 ID，我们可以清晰地重构出 Agent 的完整执行树。


• langgraph_hook.py
• 提供了TrajectoryNode，一个可以被添加到任何 LangGraph 图中的独立节点。它提供了与 ReactTrajectoryHook 类似的功能，但给予开发者更大的控制权，可以在图的任意位置手动记录状态。
  
  
• trajectory_viewer.py
• TrajectoryViewer是一个离线分析工具。它读取 JSONL 文件，使用itertools.groupby**按 ****trace_id**用 itertools.groupby **按 ****trace_id**** 对所有事件进行分组**，然后格式化输出每一次完整的交互，清晰地展示了思考链和工具调用过程。

调用过程

效果

1. 实现了完整的轨迹记录与追踪：成功构建了一个从数据捕获、处理、存储到展示的端到端轨迹系统。
2. 高度解耦和可扩展：通过TrajectoryBackend抽象，系统可以轻松适配不同的生产环境存储方案。
3. 优雅的 Trace 划分机制：通过监控用户输入来判定新 Trace 的开始，符合直觉且实现简洁，准确地将多轮对话划分为独立追踪单元。
4. 双重集成模式：提供了Hook和Node两种集成方式，兼顾了自动化和灵活性，适用于不同类型的 LangGraph 应用。
5. 简化的实现逻辑：从最初复杂的节点装饰器和状态快照方案，演进到最终“扫描消息增量”的模式，代码更简洁、鲁棒性更强。
6. 实用的可视化工具：TrajectoryViewer提供了类似 LangSmith 的分组展示功能，极大地提升了调试和分析的效率。
   
   

未来展望



虽然当前系统已功能完备，但距离真实的线上系统仍有许多可以扩展和优化的方向，
• 丰富后端支持：实现更多TrajectoryBackend，如 Kafka（用于实时数据流）、Redis（用于快速缓存）或 ClickHouse/PostgreSQL（用于持久化存储和复杂查询）。
• 增强可视化界面：将TrajectoryViewer从命令行工具升级为交互式Web 应用，提供更丰富的过滤、搜索和可视化功能。
• 与日志/数据管道集成：在生产环境中，将TrajectoryBackend对接到公司统一的日志系统和数据管道（如 Flink、Spark），实现企业级的监控和分析。
• 优化 Trace 结束判定：探索更复杂的 Trace 结束逻辑，例如基于特定事件或超时机制，以适应更复杂的业务场景。
👀
在实际的大模型工程的工作中，每一个小组件都经过多轮的评审和打磨，拿这个数据举例，可以做的事情还非常多，比如：
• 要满足数据合规的要求，有的不可以存明文，可能就需要大模型改写信息来脱敏
• 关联用户的点赞，不点赞的行为，分析上下文，来改写模型
• 将数据进行冷热分离，持久归档冷数据到便宜的存储，省钱
等等，但这些很多都是传统的后端和大数据的技术栈。有句话叫：所有的产品都值得被大模型重做一遍。同样的：所有的技术都会在大模型上找到影子。

---

欢迎光临 链载Ai (https://www.lianzai.com/)

Powered by Discuz! X3.5