企业官网建设流程全解析

面试官问：“你做过 AI Agent 项目吗？讲讲你的设计。” 你心里一紧，是不是只能回答“用 LangChain 调了调 API，做了个聊天机器人”？然后看着面试官礼貌但失望的眼神。

别慌。今天要聊的，不是一个玩具项目，而是一个能真正体现你工程化思维、可观测性设计和解决实际业务痛点的企业级 AI Agent 项目架构。它的核心不是“让 AI 回答问题”，而是让 AI 的执行过程变得透明、可控、可评估。这正是当前 AI 应用从 Demo 走向生产的关键瓶颈，也是面试中能让你脱颖而出的高价值话题。

本文将带你手把手构建一个基于LangGraph和MCP（Model Context Protocol）的 AI Agent 系统，重点实现全链路追踪（Tracing）、多维度评估（Evaluation）和系统可观测（Observability）。你会得到一套可以直接用于面试项目阐述，甚至稍作修改就能落地到真实业务中的代码框架和设计思路。我们不止讲“是什么”，更会深入“为什么重要”、“解决了什么工程问题”以及“如何避开那些坑”。

1. 这篇文章真正要解决的问题：从玩具 Demo 到生产级 Agent 的鸿沟

很多开发者学习 AI Agent 的第一个项目，往往是跟着教程，用 LangChain 或 LlamaIndex 快速拼凑一个能联网搜索、能总结文档的聊天应用。这很好，但它只是一个“玩具”。当你想把它部署到线上，服务真实用户时，一系列棘手问题会瞬间涌现：

• 黑盒与失控：Agent 内部究竟调用了哪些工具？思考步骤（Reasoning）是怎样的？为什么最终给出了这个答案？一旦出错，你几乎无从排查。
• 效果无法量化：这个 Agent 的“好坏”全凭感觉。回答得流畅就好吗？有没有“一本正经地胡说八道”？如何用数据证明它的效果在提升？
• 成本不可控：一次复杂的任务可能调用多次大模型和外部 API，成本是多少？哪些步骤最耗 Token？有没有优化空间？
• 技能难以复用与扩展：新加一个工具（如查询数据库、调用内部系统）是否需要重写大量胶水代码？如何优雅地管理越来越多的 Agent 技能？

本文要解决的，正是这些工程化问题。我们将使用LangGraph来构建具备复杂状态流转和循环能力的 Agent，用MCP协议来标准化、解耦地管理各种工具（技能），并在此基础上，搭建一套完整的追踪、评估与可观测体系。最终，你将拥有一个不仅“能跑”，而且“看得清”、“管得住”、“测得准”的 AI Agent 系统。

2. 核心概念与架构选型：为什么是 LangGraph + MCP？

在深入代码之前，必须理解我们为什么选择这个技术栈。这不仅是工具介绍，更是架构思想的体现。

2.1 LangGraph：超越链（Chain）的图（Graph）思维

• LangChain的核心是“链”（Chain），它适合线性、确定性的任务流。但对于需要根据中间结果做判断、循环、分支的复杂 Agent，链就显得力不从心。
• LangGraph是 LangChain 团队推出的用于构建有状态、多环节 Agent 的框架。它将工作流抽象为“图”（Graph），节点（Node）代表一个执行步骤（如调用 LLM、执行工具），边（Edge）代表状态流转的条件。
• 关键优势：
  • 显式状态管理：所有步骤共享并更新一个中央状态（State），数据流转清晰。
  • 支持循环与条件分支：轻松实现“思考-行动-观察”的 ReAct 循环，或根据结果选择不同路径。
  • 天然支持并发：可以定义并行执行的节点。
  • 可视化与调试：LangGraph 自带可视化工具，能直观看到执行路径，这对调试复杂 Agent 至关重要。

简单类比：如果说 LangChain 是组装一条流水线，那么 LangGraph 就是设计一张交通路线图，车辆（状态）可以根据路况（条件）选择不同的道路（边）到达各个站点（节点）。

2.2 MCP (Model Context Protocol)：工具管理的“统一插座”

• 问题：传统方式中，工具（Tool）的逻辑硬编码在 Agent 代码里。每增加一个新工具（如查天气、查股票、操作数据库），都要修改 Agent 核心代码，并处理复杂的认证、参数解析等问题。
• MCP是由 Anthropic 提出并逐渐成为行业事实标准的一个协议。它旨在将工具（或更广义的“上下文”）的提供与 AI 模型的使用进行解耦。
• 核心思想：工具以独立的MCP Server形式存在，通过标准协议（HTTP/SSE）暴露其能力。AI 应用（如我们的 LangGraph Agent）作为MCP Client，通过协议动态发现、描述并调用这些工具。
• 关键优势：
  • 解耦与复用：工具开发独立于 Agent 开发。一个查询数据库的 MCP Server 可以被公司内所有 AI 项目复用。
  • 标准化：统一的工具发现、调用和错误处理机制。
  • 安全与权限：可以在 MCP Server 层面实现精细的权限控制和审计。
  • 生态丰富：社区已经提供了大量开源的 MCP Server（用于文件系统、数据库、Git、JIRA 等）。

简单类比：MCP 就像电脑的 USB 协议。你不需要为了读U盘而去修改电脑主板（Agent），只需要找一个符合 USB 协议（MCP）的U盘（MCP Server）插上，电脑就能自动识别并使用它。

2.3 追踪（Tracing）、评估（Evaluation）与可观测性（Observability）

这是本项目的“灵魂”，也是区分玩具与生产系统的关键。

• 追踪（Tracing）：记录 Agent 执行全生命周期的详细日志。包括：每次 LLM 调用的输入/输出、每次工具调用的参数/结果、每个状态节点的流转。这解决了“黑盒”问题，是调试和优化的基础。我们将使用 LangSmith（LangChain 官方平台）来实现。
• 评估（Evaluation）：量化 Agent 的表现。不仅仅是最终答案的对错，还包括：过程评估（推理步骤是否合理）、工具使用评估（调用工具是否必要且正确）、成本评估（Token 消耗）、延迟评估等。我们将设计一套评估体系，并利用追踪数据自动执行评估。
• 可观测性（Observability）：在追踪和评估的基础上，构建仪表盘（Dashboard）和告警（Alerting），让研发和运维人员能够实时洞察系统健康度、性能瓶颈和异常情况。

我们的架构：LangGraph作为 Agent 的“大脑”和“调度中心”，MCP作为标准化、可插拔的“四肢”（工具集），而追踪与评估体系则是覆盖整个系统的“神经系统”和“体检中心”。

3. 环境准备与项目初始化

我们开始动手。请确保你的开发环境满足以下要求：

• Python 版本: 3.10 或以上（推荐 3.11+）
• 包管理工具: pip 或 poetry
• 关键 API 密钥:
  • OpenAI API Key（或其他兼容 OpenAI API 的模型服务，如 Azure OpenAI, Together.ai, 本地部署的 Ollama 等）
  • LangSmith API Key（用于追踪和评估，这是 LangChain 的官方平台，提供免费额度）

3.1 创建项目并安装依赖

首先，创建一个新的项目目录并初始化虚拟环境。

mkdir enterprise-ai-agent && cd enterprise-ai-agent python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate

创建requirements.txt文件，内容如下：

# 核心框架 langgraph==0.0.52 langchain-core==0.1.0 langchain-openai==0.0.5 langchain-mcp-adapters==0.0.3 # 用于连接MCP Server # MCP 协议相关 (以SQLite示例Server为例) mcp[cli]==1.2.0 # 安装一个社区MCP Server示例：sqlite # 通常MCP Server会单独发布，这里我们演示如何运行一个Server # 追踪与评估 langsmith==0.1.0 # 工具与工具包 python-dotenv==1.0.0 # 管理环境变量 pydantic==2.5.0 # 数据验证

安装依赖：

pip install -r requirements.txt

3.2 配置环境变量

创建.env文件，存放你的敏感信息。切记不要将此文件提交到版本控制系统（如Git）。

# .env OPENAI_API_KEY=sk-your-openai-api-key-here LANGCHAIN_API_KEY=ls-your-langsmith-api-key-here LANGCHAIN_TRACING_V2=true LANGCHAIN_PROJECT=Enterprise-AI-Agent-Demo # 在LangSmith中创建的项目名 LANGCHAIN_ENDPOINT=https://api.smith.langchain.com

在代码中，使用dotenv加载配置：

# config.py import os from dotenv import load_dotenv load_dotenv() OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") LANGCHAIN_API_KEY = os.getenv("LANGCHAIN_API_KEY") assert OPENAI_API_KEY, "OPENAI_API_KEY 未设置" assert LANGCHAIN_API_KEY, "LANGCHAIN_API_KEY 未设置"

4. 构建核心：基于 LangGraph 的 Agent

我们将构建一个具备 ReAct（Reasoning + Acting）能力的 Agent。它的任务是：理解用户问题，决定是否需要调用工具，并综合所有信息给出最终答案。

4.1 定义 Agent 状态（State）

状态是 LangGraph 中贯穿整个执行流程的数据容器。我们使用TypedDict来定义。

# agent/state.py from typing import TypedDict, List, Annotated import operator from langgraph.graph.message import add_messages class AgentState(TypedDict): """Agent 的全局状态定义""" # 消息历史，LangGraph 内置的合并操作符 `add_messages` 会处理它 messages: Annotated[List, add_messages] # 用户当前的问题 question: str # Agent 的“思考”过程（供调试和评估用） reasoning_steps: List[str] # 已调用工具的结果列表 tool_results: List[dict] # 最终答案 final_answer: str | None

Annotated和add_messages是 LangGraph 的语法糖，用于自动合并消息列表，非常方便。

4.2 创建工具（Tools）并集成 MCP Client

传统方式是将工具函数直接绑定到 LLM。我们将演示如何集成一个 MCP Client，让它动态发现和使用工具。首先，我们创建一个简单的本地工具作为备选，然后连接 MCP。

# agent/tools.py from langchain.tools import tool from datetime import datetime @tool def get_current_time(query: str) -> str: """当用户询问当前时间或日期时调用此工具。query 是用户关于时间的问题。""" now = datetime.now() return f"当前时间是：{now.strftime('%Y-%m-%d %H:%M:%S')}" # 注意：更复杂的工具（如查询数据库、调用API）我们将通过 MCP 来集成。

接下来，我们初始化 MCP Client。这里假设我们已经有一个在本地运行的 MCP Server（例如一个提供 SQLite 查询能力的 Server）。如何运行 MCP Server 将在下一节详述。

# agent/mcp_client.py import subprocess import time from typing import Any from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def create_mcp_session(server_command: list[str]) -> ClientSession: """创建并连接到一个 MCP Server""" server_params = StdioServerParameters(command=server_command) stdio_transport = await stdio_client(server_params) session = ClientSession(*stdio_transport) await session.initialize() return session async def get_mcp_tools(session: ClientSession) -> list[dict[str, Any]]: """从 MCP Server 获取工具列表""" # 调用 MCP 标准接口 `list_tools` response = await session.list_tools() # 将 MCP 工具描述转换为 LangChain Tool 格式 tools = [] for tool_info in response.tools: # 这里需要根据 MCP 返回的格式做适配转换 # 简化示例：假设 tool_info 有 name, description, inputSchema langchain_tool = { "name": tool_info.name, "description": tool_info.description, "args_schema": ... # 根据 inputSchema 创建 Pydantic 模型 } tools.append(langchain_tool) return tools

4.3 定义 LangGraph 节点（Nodes）和边（Edges）

这是 Agent 的逻辑核心。我们创建几个关键节点：

1. reasoning_node: 让 LLM 分析问题，决定下一步是调用工具还是直接回答。
2. tool_node: 执行被选中的工具。
3. final_answer_node: 生成最终答案。

# agent/graph.py from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from .state import AgentState from .tools import get_current_time import json # 初始化 LLM llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) # 使用成本更低的模型进行推理 # 绑定工具（这里先绑定本地工具，后续会替换为从MCP动态获取的工具） tools = [get_current_time] llm_with_tools = llm.bind_tools(tools) def reasoning_node(state: AgentState) -> dict: """推理节点：分析问题，决定行动""" messages = state["messages"] question = state["question"] # 构造给 LLM 的提示 system_prompt = """你是一个有帮助的AI助手。请仔细思考用户的问题。 如果你有合适的工具可以获取更准确的信息，请调用工具。 如果你已经拥有足够的信息来回答问题，请直接给出答案。""" prompt = [{"role": "system", "content": system_prompt}, {"role": "user", "content": question}] # 调用 LLM response = llm_with_tools.invoke(prompt) # 记录推理步骤 reasoning_steps = state.get("reasoning_steps", []) reasoning_steps.append(f"思考: {response.content}") # 检查 LLM 是否想调用工具 tool_calls = response.tool_calls if hasattr(response, 'tool_calls') else [] new_state = { "messages": messages + [response], "reasoning_steps": reasoning_steps, } if tool_calls: # 下一步去执行工具 new_state["next"] = "call_tool" new_state["tool_calls"] = tool_calls else: # 下一步去生成最终答案 new_state["next"] = "final_answer" new_state["draft_answer"] = response.content return new_state def tool_node(state: AgentState) -> dict: """工具执行节点""" tool_calls = state["tool_calls"] tool_results = state.get("tool_results", []) for tc in tool_calls: tool_name = tc["name"] tool_args = tc["args"] # 在实际项目中，这里应该是一个工具名称到可调用对象的映射 # 我们简化处理，只处理我们已知的工具 if tool_name == "get_current_time": result = get_current_time.invoke(json.dumps(tool_args)) else: result = f"错误：未知工具 {tool_name}" # 记录结果 tool_results.append({ "tool_name": tool_name, "args": tool_args, "result": result }) # 构造 ToolMessage 供 LLM 理解 tool_message = ToolMessage(content=str(result), tool_call_id=tc.get("id", "")) state["messages"].append(tool_message) new_state = { "tool_results": tool_results, "next": "reasoning" # 执行完工具后，回到推理节点进行下一步分析 } return new_state def final_answer_node(state: AgentState) -> dict: """最终答案生成节点""" draft_answer = state.get("draft_answer", "") question = state["question"] # 这里可以引入一个“总结”LLM调用，整合所有工具结果和历史消息，生成最终答案。 # 为了简化，我们直接使用草稿答案或基于所有信息再生成一次。 final_messages = state["messages"] # 添加一个系统提示，要求生成友好、准确的最终答案 final_prompt = final_messages + [HumanMessage(content=f"基于以上对话和信息，请直接给出问题'{question}'的最终答案。")] final_response = llm.invoke(final_prompt) new_state = { "final_answer": final_response.content, "next": END # 指向结束 } return new_state

现在，我们用这些节点构建图（Graph）：

# agent/graph.py (续) def create_agent_graph(): """创建并编译 Agent 工作流图""" workflow = StateGraph(AgentState) # 添加节点 workflow.add_node("reasoning", reasoning_node) workflow.add_node("call_tool", tool_node) workflow.add_node("final_answer", final_answer_node) # 设置入口点 workflow.set_entry_point("reasoning") # 添加边（根据状态中的 `next` 字段决定流向） workflow.add_conditional_edges( "reasoning", # 这个函数根据 state 返回下一个节点的名称 lambda state: state.get("next", "final_answer"), { "call_tool": "call_tool", "final_answer": "final_answer" } ) workflow.add_edge("call_tool", "reasoning") # 工具执行完回到推理 workflow.add_edge("final_answer", END) # 编译图 app = workflow.compile() return app # 创建图应用实例 agent_app = create_agent_graph()

5. 集成 MCP Server：以 SQLite 查询为例

让我们的 Agent 拥有查询数据库的能力。我们将使用一个社区提供的 SQLite MCP Server。

5.1 启动 MCP Server

首先，确保你安装了mcpCLI。然后，我们可以通过 Python 脚本或命令行启动一个 Server。这里我们使用mcp包内置的示例 Server，或者使用社区的sqliteserver。

假设我们有一个chinook.db示例数据库。创建一个mcp_servers/sqlite_server.py脚本：

# mcp_servers/sqlite_server.py import asyncio from mcp.server import Server, StdioServerParameters from mcp.server.models import Tool, TextContent import sqlite3 import json from pathlib import Path # 创建一个简单的 SQLite 工具 Server server = Server("sqlite-server") @server.list_tools() async def handle_list_tools() -> list[Tool]: return [ Tool( name="query_sqlite", description="执行一个 SQLite 查询语句。参数 'sql' 必须是合法的 SQL SELECT 语句。", inputSchema={ "type": "object", "properties": { "sql": {"type": "string", "description": "要执行的 SQL SELECT 语句"} }, "required": ["sql"] } ) ] @server.call_tool() async def handle_call_tool(name: str, arguments: dict) -> list[TextContent]: if name == "query_sqlite": db_path = Path("./chinook.db") # 你的数据库文件路径 if not db_path.exists(): return [TextContent(type="text", text=f"错误：数据库文件 {db_path} 不存在")] sql = arguments.get("sql", "") if not sql.strip().upper().startswith("SELECT"): return [TextContent(type="text", text="错误：只允许执行 SELECT 语句")] try: conn = sqlite3.connect(str(db_path)) conn.row_factory = sqlite3.Row # 返回字典样式的行 cursor = conn.cursor() cursor.execute(sql) rows = cursor.fetchall() conn.close() # 将结果格式化为易读的文本 if rows: columns = [description[0] for description in cursor.description] result_text = " | ".join(columns) + "\n" + "-"*50 + "\n" for row in rows: result_text += " | ".join(str(row[col]) for col in columns) + "\n" result_text += f"\n共 {len(rows)} 行。" else: result_text = "查询成功，但未返回任何数据。" return [TextContent(type="text", text=result_text)] except sqlite3.Error as e: return [TextContent(type="text", text=f"SQLite 错误: {e}")] else: return [TextContent(type="text", text=f"未知工具: {name}")] async def main(): async with server.run_stdio_server() as (read_stream, write_stream): # 保持 Server 运行 await asyncio.Future() if __name__ == "__main__": asyncio.run(main())

运行这个 Server：

python mcp_servers/sqlite_server.py

5.2 修改 Agent 以动态加载 MCP 工具

现在，我们需要修改之前的agent/graph.py，在初始化时连接 MCP Server 并获取工具，然后绑定给 LLM。

# agent/graph.py (修改部分) import asyncio from .mcp_client import create_mcp_session, get_mcp_tools async def initialize_agent_with_mcp(): """异步初始化 Agent，连接 MCP Server""" # 1. 连接 MCP Server # 假设我们的 SQLite Server 通过 stdio 运行，命令是 `python mcp_servers/sqlite_server.py` server_command = ["python", "mcp_servers/sqlite_server.py"] session = await create_mcp_session(server_command) # 2. 获取 MCP 工具并转换为 LangChain Tool 格式 # 注意：这是一个复杂的过程，需要适配 MCP 工具描述到 LangChain Tool 对象。 # 这里我们进行大幅简化，假设我们已经转换好了一个工具列表 `mcp_tools_list`。 # 在实际项目中，你需要编写适配层代码。 from langchain.tools import StructuredTool from pydantic import BaseModel, Field # 示例：手动创建一个基于 MCP 的 Tool 包装器 class QuerySQLiteInput(BaseModel): sql: str = Field(description="要执行的 SQL SELECT 语句") async def query_sqlite_tool(sql: str) -> str: """调用 MCP Server 执行 SQL 查询""" # 通过 MCP session 调用工具 result = await session.call_tool("query_sqlite", arguments={"sql": sql}) # 提取文本内容 text_content = "" for content in result.content: if content.type == "text": text_content += content.text return text_content mcp_tool = StructuredTool.from_function( func=query_sqlite_tool, name="query_sqlite", description="执行一个 SQLite 查询语句。参数 'sql' 必须是合法的 SQL SELECT 语句。", args_schema=QuerySQLiteInput ) # 3. 组合工具（本地工具 + MCP 工具） all_tools = [get_current_time, mcp_tool] # 4. 创建绑定工具的 LLM llm_with_tools = llm.bind_tools(all_tools) # 5. 创建图（需要修改 tool_node 以支持新工具） # ... (修改 tool_node 函数，能根据名称调用对应的工具) return agent_app, session # 返回 app 和 session 以便后续关闭 # 注意：由于涉及异步，运行入口需要调整。通常我们会用一个主异步函数来运行。

6. 实现追踪（Tracing）与评估（Evaluation）

有了可运行的 Agent，我们现在给它装上“监控系统”。

6.1 集成 LangSmith 进行自动追踪

LangChain/LangGraph 已经原生集成了 LangSmith。我们之前设置的环境变量LANGCHAIN_TRACING_V2=true和LANGCHAIN_API_KEY就是为了这个。只要使用 LangChain 的组件（如ChatOpenAI,bind_tools）和 LangGraph，调用过程会自动被记录到 LangSmith。

现在，运行你的 Agent，然后去 LangSmith 官网 查看你的项目Enterprise-AI-Agent-Demo，你就能看到每一次运行的详细追踪信息，包括每个节点的输入输出、耗时、Token 使用情况等。

6.2 设计并实现评估体系（Evaluation）

追踪记录了“发生了什么”，评估则要判断“做得好不好”。我们设计几个评估维度：

1. 最终答案相关性（Answer Relevance）：最终答案是否直接回答了用户问题？
2. 工具调用合理性（Tool Use Appropriateness）：调用的工具是否必要？参数是否正确？
3. 推理过程连贯性（Reasoning Coherence）：思考步骤是否逻辑清晰？
4. 成本与延迟（Cost & Latency）：本次调用消耗了多少 Token？总耗时多少？

我们将编写评估函数，并利用 LangSmith 的评估功能来批量运行和打分。

# evaluation/evaluators.py from langsmith.evaluation import evaluate, EvaluationResult from langsmith.schemas import Example, Run import asyncio def answer_relevance_evaluator(run: Run, example: Example) -> EvaluationResult: """评估最终答案的相关性""" # run.outputs 包含了 Agent 运行的最终输出 final_answer = run.outputs.get("final_answer", "") # example.outputs 是期望的输出（如果有的话），这里我们主要看输入问题 question = example.inputs.get("question", "") # 简单逻辑：如果答案长度太短或不包含问题关键词，则相关性低 # 在实际项目中，这里应该调用一个 LLM 作为评估器（LLM-as-a-judge） score = 0.0 feedback = "" if len(final_answer) < 5: score = 0.2 feedback = "答案过短，可能未充分回答问题。" elif any(keyword in question.lower() for keyword in ["时间", "日期"]) and ("时间" in final_answer or "日期" in final_answer): score = 0.9 feedback = "答案明确包含了时间信息，相关性高。" else: # 更复杂的评估可以在这里添加 score = 0.7 feedback = "答案基本相关。" return EvaluationResult(key="answer_relevance", score=score, comment=feedback) def tool_use_evaluator(run: Run, example: Example) -> EvaluationResult: """评估工具使用的合理性""" # 从 run 的中间步骤中提取工具调用信息 tool_calls = [] for step in run.child_runs or []: if step.name == "tool_node" or step.run_type == "tool": tool_calls.append(step) if not tool_calls: # 没有调用工具，评估是否应该调用 question = example.inputs.get("question", "") if any(keyword in question.lower() for keyword in ["查询", "数据库", "select"]): score = 0.3 feedback = "问题可能涉及数据查询，但未调用工具。" else: score = 0.9 feedback = "无需调用工具，判断合理。" else: # 调用了工具，评估工具选择和参数 score = 0.8 feedback = f"调用了 {len(tool_calls)} 次工具。" # 可以进一步分析具体调用了什么工具，参数是否正确 return EvaluationResult(key="tool_use_appropriateness", score=score, comment=feedback) # 可以定义更多的评估器...

然后，我们可以使用 LangSmith 的evaluate函数，在一个测试数据集上运行我们的 Agent 并自动评估。

# evaluation/run_evaluation.py from langsmith import Client from agent.graph import agent_app # 导入我们编译好的 Agent from evaluation.evaluators import answer_relevance_evaluator, tool_use_evaluator client = Client() # 1. 创建测试数据集（示例） dataset_name = "agent-qa-test" # 如果数据集不存在，则创建 try: dataset = client.read_dataset(dataset_name=dataset_name) except: dataset = client.create_dataset(dataset_name=dataset_name) # 添加一些测试用例 examples = [ ("现在几点了？", "应该调用 get_current_time 工具并返回时间。"), ("数据库里有多少首歌曲？", "应该调用 query_sqlite 工具执行 SELECT COUNT(*) FROM tracks;"), ("你好，请介绍一下你自己。", "无需调用工具，直接回答。"), ] for input_q, reference_output in examples: client.create_example( inputs={"question": input_q}, outputs={"expected": reference_output}, dataset_id=dataset.id ) # 2. 定义一个预测函数（predict function） def predict_agent(inputs: dict) -> dict: """包装我们的 Agent，使其符合 LangSmith evaluate 的接口""" question = inputs["question"] # 注意：这里需要根据你的 Agent 输入格式调整 # 假设我们的 Agent 的输入状态是 {"messages": [HumanMessage(...)], "question": question} from langchain_core.messages import HumanMessage initial_state = { "messages": [HumanMessage(content=question)], "question": question, "reasoning_steps": [], "tool_results": [], "final_answer": None } # 同步运行 Agent（如果 Agent 是异步的，需要 asyncio.run） final_state = agent_app.invoke(initial_state) return {"final_answer": final_state.get("final_answer", "No answer generated.")} # 3. 运行评估 experiment_results = evaluate( predict_agent, data=dataset_name, evaluators=[answer_relevance_evaluator, tool_use_evaluator], experiment_prefix="my-agent-eval-v1", # metadata={"version": "1.0"}, # num_repetitions=3, # 可以设置重复次数以评估稳定性 ) print(f"评估完成！查看结果：{experiment_results['project_url']}")

运行此脚本后，你可以在 LangSmith 的对应项目中看到一个详细的评估实验，包含每个测试用例的得分和反馈。

7. 构建可观测仪表盘与常见问题排查

7.1 利用 LangSmith 仪表盘

LangSmith 本身就是一个强大的可观测平台。你可以：

• 查看追踪列表：了解所有历史运行。
• 筛选与搜索：按时间、标签、状态（成功/错误）筛选。
• 深入单个追踪：查看完整的执行图、每个节点的输入输出、Token 使用、延迟。
• 查看评估结果：在“Evaluations”标签页查看整体评估指标和每个案例的详情。
• 设置告警（企业版功能）：当错误率上升或延迟超标时收到通知。

7.2 常见问题与排查思路

8. 最佳实践与工程建议

将上述组件组合成一个健壮的生产系统，还需要考虑以下几点：

1. 配置化管理：将模型类型、温度、MCP Server 地址、数据库连接串等全部抽取到配置文件（如config.yaml）或环境变量中。
2. 错误处理与回退：在每个节点（尤其是工具调用和 LLM 调用）添加try...except，并设计优雅的回退策略（例如，工具调用失败后，让 LLM 基于已有知识回答）。
3. 状态持久化：对于长对话或需要记忆的 Agent，需要将AgentState持久化到数据库（如 Redis），并在每次请求时加载。
4. 异步优化：MCP 调用、LLM 调用、数据库查询都是 I/O 密集型操作，应使用异步（async/await）来提高并发性能。确保你的整个调用链路是异步的。
5. 安全与权限：
   • MCP Server 权限：每个 MCP Server 应运行在最小权限下。数据库查询 Server 应使用只读账号。
   • 输入验证与清理：对用户输入和 LLM 生成的工具参数进行严格的验证和清理，防止 SQL 注入等攻击。
   • 成本限制：在 Agent 层面设置最大 Token 消耗或最大循环次数，防止恶意或异常查询导致巨额费用。
6. 版本控制：对 Agent 的提示词、图结构、工具集进行版本控制。LangSmith 可以关联 Git Commit，便于追踪变更对效果的影响。
7. 评估常态化：建立自动化的评估流水线。每当 Agent 代码或提示词更新时，自动在测试数据集上运行评估，确保核心指标（答案相关性、工具使用合理率）没有下降。

9. 总结与项目延伸方向

通过本文，我们完成了一个从零到一、具备企业级特征的 AI Agent 项目。它不再是简单的提示词拼接，而是一个拥有清晰工作流（LangGraph）、标准化技能接入（MCP）、全方位可观测性（LangSmith 追踪与评估）的复杂系统。

回顾核心价值：

• 对面试官：你可以清晰地阐述 Agent 的状态机设计、工具解耦思想（MCP）、以及如何保障系统的可观测性与可评估性。这远超“我调过 API”的层面。
• 对实际项目：这套架构为 AI Agent 的迭代优化提供了数据基础（靠追踪），为效果保障建立了标准（靠评估），为功能扩展降低了成本（靠 MCP）。

你可以继续深化的方向：

1. 实现更复杂的图：引入并行节点（如同时查询多个数据源）、子图（将复杂功能模块化）、人工审核节点。
2. 集成更多 MCP Server：探索社区，接入 GitHub、JIRA、Confluence、内部 CRM 等系统的 MCP Server，快速赋予 Agent 新能力。
3. 构建自定义评估器：使用更强大的 LLM（如 GPT-4）作为裁判（LLM-as-a-judge），对答案的事实性、安全性、有帮助性进行深度评估。
4. 开发管理界面：基于追踪和评估数据，构建一个内部仪表盘，让产品经理和运营也能直观看到 Agent 的性能和瓶颈。
5. 探索本地模型：将核心 LLM 替换为 Ollama 运行的本地模型（如 Llama 3.2, Qwen2.5），在保证效果的同时控制成本和数据隐私。

这个项目框架为你打开了一扇门，门后是构建可靠、可扩展、可评估的智能应用的全新工程范式。建议你克隆代码，从连接一个真实的数据库 MCP Server 开始，亲手体验从“黑盒”到“白盒”的掌控感。在面试中，这份经历和思考，将成为你最有力的证明。