在实际 AI Agent 项目开发中很多开发者能快速搭建一个基于大语言模型的对话原型但一旦涉及多步骤决策、状态管理、工具调用追踪和效果评估项目就会变得难以维护和观测。面试官追问的“如何设计一个健壮的 Agent 系统”或“如何追踪和评估 Agent 的执行链路”往往成为区分普通应用和可交付企业级项目的关键。本文将围绕 LangGraph 和 MCP 这两个核心工具构建一个具备完整执行追踪、效果评估和系统可观测性的 AI Agent 项目。通过这个项目你将理解如何将一个简单的提示词工程升级为一个结构清晰、状态可控、便于调试和迭代的生产级智能体系统。1. 理解 AI Agent 的核心挑战与 LangGraph 的解决方案一个基础的 AI Agent 通常由“思考-行动-观察”循环构成。然而当任务变得复杂需要串联多个工具、处理分支逻辑或维护长期记忆时仅靠简单的循环和提示词会迅速导致代码混乱、状态丢失和问题难以排查。1.1 传统 Agent 实现的主要痛点在简单的循环结构中开发者通常面临几个核心问题状态管理困难Agent 的思考上下文、历史对话、工具调用结果等状态分散在各个变量或临时存储中难以持久化和回溯。执行流不透明Agent 内部如何决策、调用了哪个工具、返回了什么结果这一系列步骤缺乏清晰的链路记录如同黑盒。缺乏结构化控制难以实现复杂的流程控制如条件分支、循环、并行执行或人工审批节点。评估与调试成本高当 Agent 输出不符合预期时没有系统的日志和追踪信息来定位问题发生在“思考”、“工具调用”还是“结果解析”环节。1.2 LangGraph 的设计哲学与核心概念LangGraph 是 LangChain 框架中用于构建有状态、多环节工作流的库。它将 Agent 的执行过程抽象为一个有向图其中节点代表执行步骤如调用 LLM、执行工具边代表步骤之间的流转条件。State状态一个贯穿整个图执行过程的共享字典。它定义了工作流需要维护的所有信息例如用户输入、LLM 消息历史、工具调用结果、中间结论等。状态是工作流“记忆”的载体。Node节点一个执行单元通常是一个函数。它接收当前状态执行操作如调用模型、运行代码并返回一个更新后的状态。Edge边决定执行流程如何从一个节点跳转到下一个节点。边可以是固定的always也可以是基于条件的conditional根据状态中的某个值来决定下一步走向。Graph图由节点和边组成的完整工作流定义。这种图结构使得复杂的、非线性的 Agent 逻辑变得可视化、可管理和可追踪。每个节点的输入输出都是明确的状态变更为后续的链路追踪和评估打下了坚实基础。1.3 MCP 在可观测性中的角色MCP 通常指Model Context Protocol或是在特定上下文中指代监控、追踪协议。在 AI Agent 的语境下我们将其引申为构建可观测性体系的关键。一个可观测的系统需要具备日志、指标和追踪三大支柱。日志记录离散事件如“调用了天气查询工具参数为北京”。指标聚合数据如“过去一小时工具调用成功率为 95%”。追踪记录单个请求在整个分布式系统中的完整路径在 Agent 中即一个任务从开始到结束流经了哪些节点每个节点的耗时和状态。LangGraph 的图执行天然适合注入追踪点。我们可以在每个节点的前后、每条边的判断处将执行信息节点名、输入状态、输出状态、耗时、错误发送到日志系统或专门的追踪后端从而实现对整个 Agent 决策和执行链路的透明化观测。2. 项目环境准备与核心依赖配置我们将构建一个具备文档查询与总结能力的 Agent。它需要理解用户问题从向量数据库中检索相关文档片段然后进行总结回答。整个过程将被 LangGraph 管理并集成追踪和评估模块。2.1 环境与工具清单确保你的开发环境满足以下要求组件要求说明Python3.8核心开发语言。包管理pip 或 poetry推荐使用虚拟环境。LLM 服务OpenAI API 或本地模型本文以 OpenAI GPT-4 为例也可使用 Ollama 运行本地模型。向量数据库Chroma轻量级易于本地实验。生产环境可考虑 Weaviate, Pinecone 等。文档加载LangChain 文档加载器用于加载和切割文档。追踪后端LangSmithLangChain 官方可观测性平台提供免费额度非常适合开发和评估。也可集成自定义日志。2.2 依赖安装创建一个新的项目目录并初始化虚拟环境然后安装核心依赖。# 创建项目目录并进入 mkdir ai-agent-with-langgraph cd ai-agent-with-langgraph # 创建并激活虚拟环境 (以 conda 为例) conda create -n langgraph-agent python3.10 conda activate langgraph-agent # 安装核心依赖 pip install langgraph langchain langchain-openai langchain-chroma pip install pypdf # 用于读取PDF文档 pip install python-dotenv # 用于管理环境变量如果你计划使用 LangSmith 进行追踪和评估还需要安装其 SDK 并配置 API 密钥。pip install langsmith2.3 环境变量配置在项目根目录创建.env文件用于安全存储敏感信息。# .env 文件内容 OPENAI_API_KEYsk-your-openai-api-key-here # LangSmith 配置 (可选但强烈推荐用于追踪) LANGSMITH_API_KEYls-your-langsmith-api-key-here LANGSMITH_PROJECTlanggraph-agent-tutorial LANGSMITH_TRACINGtrue在你的 Python 脚本或 Jupyter Notebook 开头加载这些环境变量。# config.py 或主脚本开头 import os from dotenv import load_dotenv load_dotenv() OPENAI_API_KEY os.getenv(OPENAI_API_KEY) LANGSMITH_API_KEY os.getenv(LANGSMITH_API_KEY) LANGSMITH_PROJECT os.getenv(LANGSMITH_PROJECT, default-project)3. 构建基础 LangGraph Agent文档问答工作流我们首先构建一个不包含追踪和评估的基础版 Agent理解 LangGraph 的工作模式。3.1 定义共享状态状态是工作流的“记忆体”。我们定义一个 TypedDict 来明确状态的结构。from typing import TypedDict, List, Annotated import operator from langgraph.graph import StateGraph, END # 定义状态结构 class AgentState(TypedDict): question: str # 用户原始问题 documents: List[str] # 检索到的相关文档片段 answer: str # 最终生成的答案 # 用于控制流程的中间变量 retrieval_required: bool # 是否需要检索文档 # 可以扩展更多字段如聊天历史、工具调用记录等 # 注意LangGraph 推荐使用 Annotated 进行状态合并但对于简单示例我们直接使用字典更新。3.2 初始化 LLM 与工具我们使用 OpenAI 的模型并创建一个简单的“检索文档”工具这里用模拟函数代替真实的向量库查询。from langchain_openai import ChatOpenAI from langchain_core.tools import tool # 初始化 LLM llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0, api_keyOPENAI_API_KEY) # 模拟一个文档检索工具 tool def retrieve_documents(question: str) - List[str]: 根据用户问题从知识库中检索最相关的文档片段。 此处为模拟实际应连接向量数据库。 # 模拟返回一些文档片段 mock_docs [ f根据公司知识库关于{question}相关条款指出用户需提前24小时提交申请。, f技术文档记载处理此类问题通常需要检查网络配置和服务器日志。, f历史记录显示类似问题在2023年Q4曾出现解决方案是升级到v2.1版本。 ] print(f[工具调用] retrieve_documents 被调用问题: {question}) return mock_docs[:2] # 返回前两个模拟结果3.3 创建图节点Node每个节点是一个函数接收当前状态执行操作并返回更新后的状态。# 节点1路由节点 - 判断是否需要检索文档 def route_question(state: AgentState) - AgentState: 根据问题复杂度决定是否需要检索文档。 question state[question].lower() # 简单规则如果问题包含“总结”或“根据文档”则需要检索 if 总结 in question or 文档 in question or 资料 in question: state[retrieval_required] True else: # 对于简单问候等直接回答无需检索 state[retrieval_required] False print(f[路由节点] 问题: {state[question]} - 需要检索: {state[retrieval_required]}) return state # 节点2检索节点 - 调用检索工具 def retrieve_node(state: AgentState) - AgentState: 执行文档检索。 if not state.get(retrieval_required): # 如果不需要检索直接跳过文档为空 state[documents] [] return state # 调用检索工具 docs retrieve_documents.invoke({question: state[question]}) state[documents] docs print(f[检索节点] 检索到 {len(docs)} 个文档片段。) return state # 节点3生成节点 - 调用LLM生成最终答案 def generate_answer(state: AgentState) - AgentState: 基于问题和检索到的文档生成答案。 question state[question] docs state.get(documents, []) # 构建提示词 if docs: context \n\n.join(docs) prompt f请基于以下提供的参考信息回答用户的问题。如果信息不足请基于你的知识回答。 参考信息 {context} 用户问题{question} 请给出清晰、准确的答案 else: prompt f请直接回答用户的问题。 用户问题{question} 请给出友好、准确的答案 # 调用LLM response llm.invoke(prompt) state[answer] response.content print(f[生成节点] 答案已生成长度: {len(state[answer])} 字符。) return state3.4 组装图并定义执行流创建图添加节点并定义节点之间的流转逻辑。# 创建图构建器 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(router, route_question) workflow.add_node(retriever, retrieve_node) workflow.add_node(generator, generate_answer) # 设置入口点 workflow.set_entry_point(router) # 定义边路由逻辑 workflow.add_conditional_edges( router, # 下一个节点的判断函数 lambda state: retriever if state[retrieval_required] else generator, { retriever: retriever, # 如果需要检索去 retriever generator: generator, # 如果不需要直接去 generator } ) # 添加固定边 workflow.add_edge(retriever, generator) # 检索完后一定去生成答案 workflow.add_edge(generator, END) # 生成答案后结束 # 编译图 app workflow.compile()3.5 运行与验证现在我们可以运行这个 Agent 来处理不同的问题观察其执行路径。# 测试用例1需要检索的复杂问题 complex_question 请总结一下项目上线需要遵循的流程。 initial_state {question: complex_question, documents: [], answer: , retrieval_required: None} print(f\n 处理复杂问题: {complex_question} ) result app.invoke(initial_state) print(f最终答案: {result[answer][:200]}...) # 打印前200字符 # 测试用例2简单问候 simple_question 你好你是谁 initial_state_simple {question: simple_question, documents: [], answer: , retrieval_required: None} print(f\n 处理简单问题: {simple_question} ) result_simple app.invoke(initial_state_simple) print(f最终答案: {result_simple[answer]})运行上述代码你将在控制台看到类似输出清晰地展示了 Agent 在不同问题下的决策路径和执行步骤 处理复杂问题: 请总结一下项目上线需要遵循的流程。 [路由节点] 问题: 请总结一下项目上线需要遵循的流程。 - 需要检索: True [工具调用] retrieve_documents 被调用问题: 请总结一下项目上线需要遵循的流程。 [检索节点] 检索到 2 个文档片段。 [生成节点] 答案已生成长度: 345 字符。 最终答案: 根据提供的参考信息项目上线流程主要涉及申请和版本管理... 处理简单问题: 你好你是谁 [路由节点] 问题: 你好你是谁 - 需要检索: False [生成节点] 答案已生成长度: 42 字符。 最终答案: 我是一个AI助手可以帮助你解答问题...至此一个具备基础路由能力的 LangGraph Agent 已经构建完成。它的执行流是清晰、可控的。4. 集成追踪与可观测性基础 Agent 的打印日志是初级的追踪。接下来我们集成 LangSmith实现企业级可观测性。4.1 配置 LangSmith 追踪确保已设置LANGSMITH_API_KEY和LANGSMITH_PROJECT环境变量。LangChain/LangGraph 对 LangSmith 有原生支持配置非常简单。# 在主脚本开头配置 LangSmith通常在创建 LLM 和工具之前 from langsmith import Client from langchain_core.tracers import LangChainTracer # 初始化 LangSmith 客户端环境变量已配置则自动读取 client Client() # 创建追踪器 tracer LangChainTracer(project_nameLANGSMITH_PROJECT) # 在调用 invoke 时传入 callbacks 参数 print(f\n 启用 LangSmith 追踪后执行 ) result_with_trace app.invoke( initial_state, config{callbacks: [tracer]} # 传入追踪器 )执行后打开 LangSmith 官网 进入你设置的项目就能看到这次调用的完整追踪记录。你可以看到整个 Graph 的执行流程、每个节点的输入输出、工具调用的详情以及 LLM 的请求和响应。4.2 自定义节点内追踪除了框架自动记录的我们可以在关键业务逻辑点添加自定义的“Span”记录更细粒度的信息或业务指标。import time from langsmith import traceable # 使用 traceable 装饰器包装一个函数它会在 LangSmith 中创建一个独立的 span traceable(namebusiness_logic_check) def complex_business_rule_check(state: AgentState) - AgentState: 模拟一个复杂的业务规则检查并记录耗时和结果。 start_time time.time() # ... 模拟一些复杂的处理逻辑 ... time.sleep(0.1) is_compliant len(state.get(question, )) 5 # 模拟规则 end_time time.time() # 自定义元数据会记录在 Span 中 from langsmith.run_helpers import get_current_run_tree current_run get_current_run_tree() if current_run: current_run.add_metadata({compliance_check_result: is_compliant, check_duration_ms: (end_time-start_time)*1000}) state[passed_compliance_check] is_compliant return state # 将这个节点加入到图中 workflow.add_node(compliance_check, complex_business_rule_check) # ... 修改图的边在适当位置插入此节点 ...4.3 构建自定义追踪日志对于无法使用 LangSmith 的环境可以构建一个简单的日志追踪器将关键事件结构化地记录到文件或日志系统中。import json import logging from datetime import datetime class AgentTracer: def __init__(self, run_id: str): self.run_id run_id self.events [] logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s) self.logger logging.getLogger(fAgentTracer-{run_id}) def log_event(self, event_type: str, node_name: str, state_snapshot: dict, metadata: dict None): 记录一个执行事件。 event { timestamp: datetime.utcnow().isoformat(), run_id: self.run_id, event_type: event_type, # e.g., node_enter, node_exit, tool_call node: node_name, state: state_snapshot.copy(), # 注意避免记录过大或敏感数据 metadata: metadata or {} } self.events.append(event) # 同时输出到日志 self.logger.info(json.dumps(event, ensure_asciiFalse, defaultstr)) def save_trace(self, filepath: str): 将追踪记录保存到文件。 with open(filepath, w, encodingutf-8) as f: json.dump(self.events, f, ensure_asciiFalse, indent2, defaultstr) # 在节点函数中使用追踪器 def traced_retrieve_node(state: AgentState, tracer: AgentTracer) - AgentState: tracer.log_event(node_enter, retriever, state) # ... 原有的检索逻辑 ... docs retrieve_documents.invoke({question: state[question]}) state[documents] docs tracer.log_event(node_exit, retriever, state, {documents_retrieved: len(docs)}) return state通过集成 LangSmith 或自定义追踪器Agent 的执行不再是黑盒。每一次调用都产生了完整的审计日志这对于调试复杂问题、分析性能瓶颈、理解 Agent 决策过程至关重要。5. 设计并实施 Agent 评估体系追踪解决了“发生了什么”的问题评估则要解决“做得好不好”的问题。对于 AI Agent评估通常围绕回答的准确性、相关性和有用性展开。5.1 定义评估指标与评分函数首先我们需要定义如何评估一个答案。评估可以是自动化的基于规则或另一个 LLM也可以是人工的。from typing import Tuple from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser # 1. 基于规则的简单评估例如检查答案是否包含关键词 def rule_based_evaluator(question: str, answer: str, documents: List[str]) - dict: 基于简单规则的评估器。 score 0 feedback [] # 规则1答案不能为空 if not answer or answer.strip() : feedback.append(答案为空。) else: score 1 feedback.append(答案非空。) # 规则2如果提供了文档答案应提及文档中的关键实体模拟 if documents: # 假设我们从文档中提取了一个关键词“申请” if 申请 in answer: score 1 feedback.append(答案引用了文档关键信息。) else: feedback.append(答案可能未充分引用文档。) # 规则3答案长度适中示例规则 if 50 len(answer) 500: score 1 feedback.append(答案长度适中。) else: feedback.append(f答案长度({len(answer)})可能不合适。) max_score 3 normalized_score score / max_score return {score: normalized_score, feedback: feedback, max_score: max_score} # 2. 基于 LLM 的评估使用另一个 LLM 作为裁判 def create_llm_evaluator(llm): 创建一个使用 LLM 进行评分和反馈的评估链。 eval_prompt ChatPromptTemplate.from_messages([ (system, 你是一个严格的评估助手。请根据提供的参考文档和问题评估答案的质量。), (human, 问题{question} 参考文档可能为空 {documents} 待评估的答案 {answer} 请从以下维度评估每项0-5分 1. 准确性答案事实是否准确是否与参考文档一致如果提供了文档 2. 相关性答案是否直接回答了问题 3. 完整性答案是否覆盖了问题的要点 4. 清晰度答案是否表达清晰、易于理解 请先给出每一项的分数然后计算总分满分20分最后提供一段简短的改进建议。 请严格按照以下格式输出 准确性[分数] 相关性[分数] 完整性[分数] 清晰度[分数] 总分[总分] 建议[你的建议] ) ]) eval_chain eval_prompt | llm | StrOutputParser() return eval_chain # 初始化评估链 llm_evaluator create_llm_evaluator(llm)5.2 将评估节点集成到 LangGraph 中我们可以将评估作为一个独立的节点加入到工作流中或者作为一个后置处理步骤。# 节点4评估节点 def evaluate_answer(state: AgentState) - AgentState: 对生成的答案进行评估。 question state[question] answer state[answer] documents state.get(documents, []) # 方法1基于规则的评估 rule_result rule_based_evaluator(question, answer, documents) state[rule_based_score] rule_result[score] state[rule_based_feedback] rule_result[feedback] # 方法2基于LLM的评估异步或耗时可根据需要开启 try: llm_eval_input { question: question, documents: \n.join(documents) if documents else 无, answer: answer } llm_eval_result llm_evaluator.invoke(llm_eval_input) state[llm_evaluation] llm_eval_result # 可以添加解析逻辑从文本中提取分数 except Exception as e: state[llm_evaluation] f评估失败: {e} print(f[评估节点] 规则评分: {state[rule_based_score]:.2f}) return state # 将评估节点添加到图中放在生成节点之后 workflow.add_node(evaluator, evaluate_answer) # 修改边generator - evaluator - END workflow.add_edge(generator, evaluator) workflow.add_edge(evaluator, END) # 重新编译图 app_with_eval workflow.compile()5.3 批量评估与结果分析对于生产系统我们需要定期对一批测试用例进行批量评估以监控 Agent 性能的变化。import pandas as pd # 定义测试集 test_cases [ {question: 项目上线流程是什么, expected_has_doc: True}, {question: 你好, expected_has_doc: False}, {question: 请根据文档告诉我如何处理网络故障, expected_has_doc: True}, ] results [] for case in test_cases: print(f\n评估测试用例: {case[question]}) initial_state {question: case[question], documents: [], answer: , retrieval_required: None} # 执行带有评估的图 final_state app_with_eval.invoke(initial_state) # 收集结果 result_record { question: case[question], answer: final_state.get(answer, )[:100], # 截取部分 retrieval_triggered: final_state.get(retrieval_required, False), rule_score: final_state.get(rule_based_score, 0), doc_used: len(final_state.get(documents, [])) 0, } results.append(result_record) # 转换为 DataFrame 进行分析 df_results pd.DataFrame(results) print(\n 批量评估结果摘要 ) print(df_results.to_string()) print(f\n平均规则得分: {df_results[rule_score].mean():.2f})通过评估体系我们可以量化 Agent 的表现识别出它在哪些类型的问题上表现不佳例如需要检索但未触发检索或答案质量得分低从而有针对性地进行优化。6. 企业级实践常见问题排查与优化将上述组件组合后一个具备追踪和评估能力的 Agent 骨架就完成了。但在企业级应用中还需要考虑更多。6.1 常见问题与排查路径问题现象可能原因检查点与排查步骤Agent 执行结果不符合预期但无报错。1. 路由逻辑错误。2. 提示词设计不佳。3. 检索工具返回无关内容。4. LLM 理解偏差。1.检查追踪日志查看 LangSmith 或自定义日志确认执行流是否按预期经过各个节点。检查retrieval_required等状态值。2.检查节点输入输出在追踪中查看每个节点的state快照确认数据传递是否正确。3.检查工具调用确认retrieve_documents工具被调用时的输入和输出。模拟工具是否返回了合理数据4.检查 LLM 输入输出在 LangSmith 中展开generator节点查看发送给 LLM 的完整提示词和返回的原始响应。提示词是否清晰包含了上下文执行速度慢。1. LLM API 调用延迟高。2. 检索工具如向量数据库查询慢。3. 网络问题。4. 图中存在不必要的串行节点。1.分析追踪时间线LangSmith 的 Trace 视图会显示每个节点的耗时。定位耗时最长的节点。2.优化慢节点如果是 LLM考虑使用更快的模型或优化提示词减少 token。如果是检索检查向量库索引、查询语句或考虑缓存。3.并行化检查图中是否有可以并行执行的节点例如检索多个不相关的信息。LangGraph 支持并行分支。评估分数持续偏低。1. 评估标准不合理。2. Agent 核心能力不足。3. 训练数据/知识库质量差。1.复核评估函数人工检查一批低分案例看评估规则或 LLM 评估提示词是否过于严苛或偏离业务目标。2.错误分析将低分案例按类型分类如“检索失败”、“总结不全面”、“答非所问”针对每一类问题优化对应节点。3.优化知识库检查检索到的文档是否相关、准确、完整。考虑优化文档切分方式或向量化模型。6.2 生产环境最佳实践配置管理将 LLM 模型参数、工具配置、图结构参数如路由阈值外置到配置文件如 YAML或配置中心避免硬编码。错误处理与重试在图中的关键节点尤其是调用外部 API 或工具处添加健壮的错误处理。例如LLM 调用可能因网络超时失败应实现指数退避重试。from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def robust_llm_invoke(prompt): return llm.invoke(prompt)状态序列化与持久化对于长对话或长时间运行的任务需要将AgentState定期序列化到数据库或缓存中以便中断后恢复。注意清理敏感信息。版本控制对 Agent 的图定义、提示词、工具集进行版本控制。当部署新版本时可以通过 LangSmith 等工具对比新旧版本的追踪和评估结果。监控告警除了追踪还应设置关键指标如每秒查询数、平均响应时间、错误率、评估分数平均值/分布的监控看板和告警。当评估分数低于阈值或错误率飙升时及时通知。6.3 扩展方向复杂工作流尝试使用 LangGraph 的State更高级特性如支持多参与者的MessagesState或实现循环、子图、人工审核节点。真实工具集成将模拟的retrieve_documents工具替换为真实的向量数据库查询、API 调用如搜索、计算、数据库操作。长期记忆通过在图状态中维护对话历史或将历史摘要存储到外部数据库为 Agent 添加会话记忆能力。多模态能力结合 LangChain 的多模态模型处理图像、音频输入或生成图表等复杂输出。在线学习与优化根据评估结果特别是人工反馈自动调整提示词或路由策略实现 Agent 的持续迭代优化。通过 LangGraph 构建清晰的工作流通过 MCP 思想建立全面的可观测性通过系统的评估体系驱动迭代这样的 AI Agent 项目才能从演示原型走向真正可靠、可维护、可进化的企业级应用。在面试中展示出你对这套完整工程化思路的理解和实践远比仅仅调用一个 API 封装类更能体现你的深度。