在实际 AI 应用开发中我们经常听到“Agent”这个概念。它被描述为能够理解目标、使用工具、与环境交互并完成复杂任务的智能体。从年初的狂热到年中的遍地开花再到如今冷静的讨论垂直领域 Agent 的开发似乎正面临一个拐点。很多开发者发现将一个概念验证PoC级别的 Agent 真正落地到生产环境解决实际的业务问题远比想象中要困难。这背后涉及的不只是模型 API 的调用更是一整套关于架构设计、工具集成、状态管理、异常处理和成本控制的工程实践。本文旨在为那些希望深入理解并动手构建 AI Agent 的开发者提供一个清晰的工程化视角。我们将暂时搁置对市场趋势的预测聚焦于一个核心问题如何从零开始搭建一个具备基本规划、执行和反思能力的 Agent 系统。我们将以当前社区中较为流行的 LangGraph 框架为例因为它清晰地体现了 Agent 的“有状态工作流”这一核心思想。通过构建一个模拟的“技术博客选题分析 Agent”你将理解 Agent 的核心组件、工作流程并掌握将其部署运行的关键步骤。最终你会获得一个可运行、可扩展的代码基础并了解在将其推向生产环境前必须考虑的工程化问题。1. 理解 Agent 的核心从概念到架构在开始写代码之前我们必须厘清几个关键概念。很多人将“调用大模型 API”等同于构建 Agent这是一个常见的误解。Agent 的本质是一个自主的决策系统。1.1 Agent 是什么不仅仅是提示词工程一个 AI Agent 通常由以下几个核心部分组成规划Planning Agent 需要理解复杂目标并将其分解为一系列可执行的子任务或步骤。这超越了简单的单轮问答。工具使用Tool Use Agent 能够调用外部工具如搜索引擎、计算器、数据库、API来获取信息或执行操作从而突破大模型自身知识的局限。记忆Memory Agent 需要具备短期记忆当前会话的上下文和长期记忆存储历史交互、知识以保持对话的一致性和进行学习。行动Action 基于规划、工具和记忆Agent 执行具体的操作可能是生成一段文本也可能是调用一个工具。反思Reflection 高级的 Agent 能够评估自身行动的结果判断任务是否完成或是否需要调整策略。这构成了一个“观察-思考-行动”的循环。因此构建 Agent 是一个系统工程它涉及提示词设计、工作流编排、状态管理、工具集成和异常处理等多个层面。1.2 为什么需要 Agent 框架LangGraph 的角色当你试图用代码实现上述循环时很快就会遇到挑战如何管理复杂的执行状态如何处理分支和循环如何让多个 Agent 协作手动用if-else和循环来控制流程会变得极其臃肿且难以维护。这就是 Agent 框架的价值所在。LangGraph是 LangChain 生态系统中的一个库它允许你将 Agent 系统建模为一个有向图Graph。图中的节点Node代表一个执行步骤如调用 LLM、运行工具边Edge代表步骤之间的流转条件。这种抽象完美地契合了 Agent 的“规划-执行-评估”循环。使用 LangGraph你可以清晰定义工作流 将 Agent 的逻辑可视化为一幅图便于理解和调试。管理复杂状态 框架帮你维护一个共享的状态字典在不同节点间传递和更新信息。处理循环和条件分支 轻松实现“直到任务完成才停止”或“根据结果选择不同路径”的逻辑。支持多 Agent 协作 可以将不同的节点分配给不同特化的 Agent如一个负责搜索一个负责写作。接下来我们将通过一个具体案例展示如何使用 LangGraph 构建一个 Agent。2. 环境准备与项目初始化我们将构建一个“技术博客选题分析 Agent”。它的目标是给定一个宽泛的技术领域如“云原生”Agent 能自动分析当前趋势生成几个具体的博客选题建议并为每个选题撰写一个简短的提纲。2.1 技术栈与依赖选择这个项目主要依赖 Python 和以下库LangChain / LangGraph: 用于构建 Agent 工作流。OpenAI API(或其它兼容 API): 为大模型提供能力。我们将使用gpt-4o-mini或gpt-4-turbo作为“大脑”。Tavily Search API: 一个为 AI 优化的搜索引擎工具用于获取实时网络信息。你也可以用 SerperAPI 或自己封装 Google Search。环境管理工具: 推荐使用conda或venv。以下是具体的依赖版本requirements.txtlangchain0.2.0 langchain-openai0.1.0 langchain-community0.0.10 langgraph0.0.52 tavily-python0.3.0 python-dotenv1.0.0注意LangChain 生态版本迭代较快以上版本为撰写时的稳定版本。实际安装前建议查看官方文档确认最新兼容版本。2.2 项目结构与关键文件创建一个清晰的项目目录结构是良好工程实践的开始。tech_blog_agent/ ├── .env # 存储API密钥等敏感信息 ├── requirements.txt # 项目依赖 ├── main.py # Agent工作流主入口 ├── agents/ │ ├── __init__.py │ ├── blog_agent.py # 定义Agent状态图和节点 │ └── tools.py # 定义自定义工具如搜索 ├── config/ │ └── settings.py # 配置管理模型、API端点等 └── run_agent.py # 运行脚本首先设置环境变量。在项目根目录创建.env文件# .env OPENAI_API_KEYsk-your-openai-api-key-here TAVILY_API_KEYyour-tavily-api-key-here重要 务必在.gitignore中加入.env避免将密钥提交到代码仓库。然后创建config/settings.py来集中管理配置# config/settings.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 class Settings: OPENAI_API_KEY os.getenv(OPENAI_API_KEY) TAVILY_API_KEY os.getenv(TAVILY_API_KEY) # 模型配置 LLM_MODEL gpt-4o-mini # 也可用 gpt-4-turbo LLM_TEMPERATURE 0.1 # 较低的温度使输出更稳定 settings Settings()3. 构建核心 Agent 工作流我们将使用 LangGraph 的“状态图StateGraph”来定义 Agent。核心是定义一个共享的“状态”和一系列操作该状态的“节点”。3.1 定义 Agent 的状态State状态是一个字典包含了工作流执行过程中需要传递和更新的所有信息。我们为博客选题 Agent 设计如下状态# agents/blog_agent.py from typing import TypedDict, List, Annotated import operator from langgraph.graph.message import add_messages from langchain_core.messages import BaseMessage, HumanMessage class AgentState(TypedDict): # 输入用户最初的问题 input: str # 消息历史用于记录与LLM的对话 messages: Annotated[List[BaseMessage], add_messages] # 规划阶段产生的子任务列表 sub_tasks: List[str] # 搜索工具返回的原始资料 search_results: str # 生成的最终博客选题列表 blog_topics: List[dict] # 当前执行到了哪个步骤用于控制流程 current_step: strAnnotated[List[BaseMessage], add_messages]是 LangGraph 的语法糖它自动帮我们处理消息列表的追加非常方便。current_step字段将用于控制图的走向例如判断是否需要进行搜索。3.2 创建工具Tools工具是 Agent 的手臂。我们首先创建一个搜索工具用于获取最新的技术趋势信息。# agents/tools.py from langchain_community.tools.tavily_search import TavilySearchResults from config.settings import settings def get_search_tool(): 初始化并返回Tavily搜索工具 tool TavilySearchResults( api_keysettings.TAVILY_API_KEY, max_results3, # 每次搜索返回3条结果避免信息过载 search_depthadvanced, # 获取更详细的内容 ) return tool # 可以在这里定义更多工具比如获取GitHub趋势、查询技术文档等。3.3 定义工作流节点Nodes节点是图中的一个执行单元。我们将工作流分解为四个主要节点规划、研究、生成和路由。节点1规划节点plan_node这个节点接收用户输入让 LLM 分析并制定执行计划子任务。# agents/blog_agent.py from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from config.settings import settings llm ChatOpenAI(modelsettings.LLM_MODEL, temperaturesettings.LLM_TEMPERATURE) def plan_node(state: AgentState): 分析用户输入规划执行步骤 prompt ChatPromptTemplate.from_messages([ (system, 你是一个资深技术博客策划专家。你的任务是将用户宽泛的请求分解为具体的、可执行的研究步骤。 例如用户输入“云原生”你可能需要研究1. 最新的云原生技术趋势如服务网格、Serverless。2. 开发者常见的痛点。3. 社区热门讨论话题。 请只输出一个步骤列表每个步骤用一句话描述。), (human, 用户请求{input}), ]) chain prompt | llm # 调用LLM获取规划结果 plan_result chain.invoke({input: state[input]}) # 假设LLM返回的是文本我们按行解析成任务列表实际项目需要更鲁棒的解析 tasks [line.strip() for line in plan_result.content.split(\n) if line.strip()] # 更新状态 new_state { **state, sub_tasks: tasks, current_step: research, # 规划完成后进入研究阶段 messages: state[messages] [HumanMessage(contentf规划完成。子任务{tasks})], } return new_state节点2研究节点research_node此节点执行规划中的第一个研究性子任务例如搜索并保存结果。# agents/blog_agent.py from agents.tools import get_search_tool search_tool get_search_tool() def research_node(state: AgentState): 执行研究任务例如搜索 if not state[sub_tasks]: # 如果没有子任务直接跳过研究 return {**state, search_results: 无研究任务, current_step: generate} # 这里简化处理只取第一个子任务进行搜索 research_query state[sub_tasks][0] # 调用搜索工具 try: results search_tool.invoke(research_query) # Tavily 返回的是字典列表我们将其转换为易读的文本 result_text \n---\n.join([ f标题{r[title]}\n内容{r[content]}\n链接{r[url]} for r in results ]) except Exception as e: result_text f搜索工具调用失败{str(e)} new_state { **state, search_results: result_text, current_step: generate, # 研究完成后进入生成阶段 messages: state[messages] [HumanMessage(contentf研究完成。查询{research_query})], } return new_state节点3生成节点generate_node基于用户输入和研究结果生成具体的博客选题和提纲。# agents/blog_agent.py def generate_node(state: AgentState): 生成博客选题和提纲 prompt ChatPromptTemplate.from_messages([ (system, 你是一位经验丰富的技术作者。基于用户的需求和提供的研究资料生成3个具体、有价值、可落地的技术博客选题。 对于每个选题请提供 1. 标题吸引人且包含核心关键词。 2. 目标读者例如初级开发者、架构师、运维工程师。 3. 核心要点3-5个文章主要部分或论点。 4. 难点与亮点这篇文章写作的挑战和独特价值在哪里。 请以清晰的JSON格式输出包含一个“topics”数组每个元素是一个包含上述字段的对象。), (human, 用户原始需求{input} 研究资料{research} 请基于以上信息生成博客选题。 ), ]) chain prompt | llm response chain.invoke({ input: state[input], research: state.get(search_results, 无可用资料) }) # 这里需要解析LLM返回的JSON。实际项目中应使用LLM的JSON模式或后处理解析。 # 为简化我们假设LLM返回了格式良好的JSON字符串并用eval演示生产环境应用json.loads并加异常处理。 import json try: # 尝试从响应文本中提取JSON部分 import re json_match re.search(rjson\n(.*?)\n, response.content, re.DOTALL) if json_match: json_str json_match.group(1) else: json_str response.content topics_data json.loads(json_str) blog_topics topics_data.get(topics, []) except (json.JSONDecodeError, AttributeError) as e: blog_topics [{error: f解析LLM输出失败{str(e)}, raw_output: response.content[:200]}] new_state { **state, blog_topics: blog_topics, current_step: end, # 生成完成后工作流结束 messages: state[messages] [HumanMessage(content选题生成完成。)], } return new_state节点4路由节点router_node这是一个条件判断节点根据当前状态决定下一步走向哪个节点。# agents/blog_agent.py def router_node(state: AgentState) - str: 根据当前步骤决定下一个节点 current state.get(current_step, plan) if current plan: return plan elif current research: return research elif current generate: return generate else: return __end__ # LangGraph 内置的特殊节点表示结束3.4 组装成图Graph将节点和边组装起来形成一个完整的工作流。# agents/blog_agent.py from langgraph.graph import StateGraph, END def create_blog_agent_workflow(): 创建并返回博客选题Agent的工作流图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(plan, plan_node) workflow.add_node(research, research_node) workflow.add_node(generate, generate_node) # 设置入口点 workflow.set_entry_point(plan) # 添加条件边。router_node 函数决定下一个节点是谁。 workflow.add_conditional_edges( plan, router_node, { plan: plan, # 理论上不会发生仅为演示 research: research, generate: generate, __end__: END, } ) workflow.add_conditional_edges( research, router_node, { research: research, generate: generate, __end__: END, } ) workflow.add_conditional_edges( generate, router_node, { generate: generate, __end__: END, } ) # 编译图 app workflow.compile() return app4. 运行与验证现在我们已经有了一个完整的 Agent 工作流。接下来创建一个主程序来运行它并查看结果。4.1 创建运行脚本# run_agent.py from agents.blog_agent import create_blog_agent_workflow, AgentState from langchain_core.messages import HumanMessage import asyncio async def main(): # 1. 初始化工作流 app create_blog_agent_workflow() # 2. 准备初始状态 user_input 云原生监控 # 可以替换为任何技术主题如“大模型微调”、“React性能优化” initial_state: AgentState { input: user_input, messages: [HumanMessage(contentuser_input)], sub_tasks: [], search_results: , blog_topics: [], current_step: plan, } # 3. 运行工作流 print(f开始执行Agent分析主题{user_input}...) print(- * 50) final_state None # 我们可以流式输出每个步骤的结果便于观察 async for event in app.astream(initial_state, stream_modevalues): node_name list(event.keys())[0] state event[node_name] print(f[节点{node_name}] 当前步骤{state.get(current_step)}) if node_name plan and state.get(sub_tasks): print(f 规划的子任务{state[sub_tasks]}) elif node_name research and state.get(search_results): # 搜索结果可能很长只打印前500字符 preview (state[search_results][:500] ...) if len(state[search_results]) 500 else state[search_results] print(f 研究结果预览{preview}) elif node_name generate and state.get(blog_topics): print(f 生成的博客选题) for i, topic in enumerate(state[blog_topics], 1): print(f 选题{i}: {topic.get(title, N/A)}) print(- * 30) final_state state # 4. 输出最终结果 print(\n *50) print(最终结果) if final_state and final_state.get(blog_topics): for i, topic in enumerate(final_state[blog_topics], 1): print(f\n选题 {i}:) print(f 标题{topic.get(title)}) print(f 目标读者{topic.get(target_audience)}) print(f 核心要点{topic.get(key_points)}) print(f 难点与亮点{topic.get(challenges_and_highlights)}) else: print(未生成有效选题。) print(f最终状态{final_state}) if __name__ __main__: asyncio.run(main())4.2 执行与输出分析在终端运行python run_agent.py。你应该能看到类似以下的输出具体内容因模型和搜索实时结果而异开始执行Agent分析主题云原生监控... -------------------------------------------------- [节点plan] 当前步骤research 规划的子任务[1. 调研当前云原生监控领域的主流工具和框架如Prometheus, Grafana, Jaeger等。, 2. 分析在微服务、容器化环境下监控面临的独特挑战如动态性、多维度指标。, 3. 收集开发者/运维在云原生监控实践中的常见痛点与最佳实践。] ------------------------------ [节点research] 当前步骤generate 研究结果预览标题2024年云原生监控全景图从Prometheus到OpenTelemetry 内容随着Kubernetes和微服务的普及监控体系正从传统基础设施监控转向可观测性...后略 链接https://example.com/article1 --- 标题微服务监控的五大挑战与解决方案 内容动态扩缩容、服务间调用链追踪、日志聚合...后略 链接https://example.com/article2 ... ------------------------------ [节点generate] 当前步骤end 生成的博客选题 选题1: 《云原生可观测性入门搭建你的第一个PrometheusGrafana监控栈》 选题2: 《告别“监控孤岛”基于OpenTelemetry实现统一链路追踪》 选题3: 《Kubernetes集群监控实战核心指标解读与告警配置优化》 最终结果 选题 1: 标题《云原生可观测性入门搭建你的第一个PrometheusGrafana监控栈》 目标读者刚接触云原生的开发者和运维人员 核心要点1. 云原生监控与传统监控的区别。 2. Prometheus数据模型与核心组件。 3. 使用Helm在K8s中快速部署。 4. Grafana可视化仪表盘配置。 5. 编写第一条PromQL查询与告警规则。 难点与亮点难点在于理解Prometheus的拉取模型和指标类型亮点是提供从零到一的完整可运行示例。 选题 2: 标题《告别“监控孤岛”基于OpenTelemetry实现统一链路追踪》 目标读者中高级后端架构师和SRE 核心要点1. 分布式追踪的核心概念Trace, Span。 2. OpenTelemetry架构与数据收集流程。 3. 在Go/Java服务中集成OTel SDK。 4. 将追踪数据导出到Jaeger/Tempo。 5. 利用追踪数据进行性能瓶颈分析。 难点与亮点难点是多语言SDK的配置差异和数据关联亮点是打通Metrics、Logs、Traces的实践方案。 选题 3: 略这个输出表明你的 Agent 已经成功运行。它完成了“规划 - 研究 - 生成”的完整循环并输出了结构化的结果。5. 关键配置、参数与工程化考量一个能跑的 Demo 和一個健壯的生产级 Agent 之间存在巨大差距。以下是几个关键的工程化考量点。5.1 模型与参数配置模型的选择和参数调优直接影响 Agent 的稳定性、成本和质量。参数推荐值说明影响模型选择gpt-4o-mini/gpt-4-turbogpt-4o-mini成本低、速度快适合简单任务和频繁调用。gpt-4-turbo推理能力更强适合复杂规划和分析。成本、响应速度、任务完成质量。Temperature0.1 - 0.3控制输出的随机性。对于需要稳定、结构化输出的 Agent 任务应设置较低的值。值越高输出越有创造性但可能不稳定值越低输出越确定但可能死板。Max Tokens根据任务设定限制单次响应的最大长度。对于规划、总结等任务可设小些如500对于生成长文需设大如2000。防止响应过长控制单次调用成本。Top P0.9 - 1.0与 Temperature 类似控制采样范围。通常与 Temperature 配合使用保持默认即可。影响输出的多样性和可预测性。在config/settings.py中集中管理这些参数# config/settings.py (扩展) class Settings: # ... 其他配置 ... LLM_MODEL gpt-4o-mini LLM_TEMPERATURE 0.1 LLM_MAX_TOKENS 1000 LLM_TOP_P 1.0 # 流式输出开关用于需要实时展示的场景 LLM_STREAMING False5.2 工具调用的错误处理与降级工具调用如网络搜索是 Agent 失败的主要风险点。必须实现健壮的错误处理。# agents/tools.py (增强版) import asyncio from typing import Optional from langchain_core.tools import ToolException def safe_tool_call(tool, input_str, max_retries2, timeout10): 带重试和超时的安全工具调用 for attempt in range(max_retries): try: # 使用异步或线程池防止阻塞这里简化用同步调用加超时 # 实际项目中如果框架支持应使用异步工具调用。 result tool.invoke(input_str) return result except ToolException as e: print(f工具调用失败 (尝试 {attempt1}/{max_retries}): {e}) if attempt max_retries - 1: raise asyncio.sleep(1) # 简单等待后重试 except Exception as e: # 网络超时、API限制等通用异常 print(f工具调用发生未知错误: {e}) return f工具暂时不可用{str(e)} return 工具调用失败请稍后重试。 # 在research_node中调用 # result_text safe_tool_call(search_tool, research_query)5.3 状态管理与持久化LangGraph 的内存状态在进程结束后就消失了。对于需要长期对话或异步执行的任务必须将状态持久化。# 示例使用数据库如SQLite持久化状态 import sqlite3 import json from datetime import datetime class StateManager: def __init__(self, db_pathagent_state.db): self.conn sqlite3.connect(db_path) self._init_db() def _init_db(self): cursor self.conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS workflow_state ( session_id TEXT PRIMARY KEY, state_json TEXT NOT NULL, updated_at TIMESTAMP NOT NULL ) ) self.conn.commit() def save_state(self, session_id: str, state: dict): cursor self.conn.cursor() state_json json.dumps(state) cursor.execute( INSERT OR REPLACE INTO workflow_state (session_id, state_json, updated_at) VALUES (?, ?, ?) , (session_id, state_json, datetime.utcnow())) self.conn.commit() def load_state(self, session_id: str) - Optional[dict]: cursor self.conn.cursor() cursor.execute(SELECT state_json FROM workflow_state WHERE session_id?, (session_id,)) row cursor.fetchone() if row: return json.loads(row[0]) return None # 在运行工作流时可以加载和保存状态 # state_manager StateManager() # saved_state state_manager.load_state(user_session_id) # if saved_state: # initial_state.update(saved_state) # final_state app.invoke(initial_state) # state_manager.save_state(user_session_id, final_state)6. 常见问题排查与调试在开发 Agent 过程中你会遇到各种问题。以下是一个快速排查清单。6.1 Agent 流程卡住或陷入循环现象可能原因检查与解决工作流一直在plan和research节点间循环。router_node逻辑错误或current_step状态更新不正确。1. 在router_node和每个节点函数内打印state[‘current_step’]。2. 确保每个节点执行后都正确更新了current_step。3. 检查条件边的映射关系是否正确。LLM 输出不符合预期导致后续节点解析失败。提示词Prompt不够清晰或 Temperature 设置过高。1. 在 Prompt 中明确指定输出格式如“请输出 JSON”。2. 使用 LLM 的 JSON 模式如response_format{“type”: “json_object”}。3. 将 Temperature 调低至 0.1。4. 在代码中添加对 LLM 输出的后处理如正则提取、重试。工具调用超时或返回错误。网络问题、API 密钥无效、工具参数错误。1. 为工具调用添加超时和重试机制见 5.2 节。2. 检查环境变量是否正确加载。3. 单独测试工具调用是否正常print(search_tool.invoke(“test”))。6.2 输出质量低下现象可能原因检查与解决生成的选题过于宽泛或空洞。1. 用户输入太宽泛。2. 搜索工具返回的结果质量差。3. 生成节点的 Prompt 约束力不够。1. 引导用户输入更具体的问题或在规划节点让 LLM 先澄清需求。2. 优化搜索查询词或更换/组合多个搜索工具。3. 在生成 Prompt 中提供更具体的范例Few-shot Prompting。选题之间重复度高。LLM 在生成多个条目时缺乏多样性。1. 在 Prompt 中明确要求“角度不同”、“覆盖不同子领域”。2. 让生成节点分两次或多次调用 LLM每次聚焦一个子方向。3. 在生成后添加一个“去重/排序”节点。6.3 性能与成本问题现象可能原因检查与解决单次请求耗时过长30秒。1. 串行调用工具和 LLM。2. 网络延迟高。3. 使用了速度慢的大模型如 GPT-4。1. 分析工作流将可以并行的节点并行化LangGraph 支持。2. 为网络请求设置合理超时并使用更稳定的工具提供商。3. 在非关键路径使用更快、更便宜的模型如用gpt-4o-mini做规划用gpt-4-turbo做最终生成。API 调用费用飙升。1. 工作流设计复杂调用链路过长。2. 每次调用都使用高 token 数的模型。3. 没有缓存重复内容。1. 优化工作流减少不必要的 LLM 调用例如缓存规划结果。2. 在config/settings.py中为不同节点配置不同的模型。3. 对搜索等工具的结果进行去重和缓存避免对相同查询重复调用。7. 从 Demo 到生产最佳实践与扩展方向要让这个 Agent 真正可用还需要在以下几个方面进行强化。7.1 增强记忆与上下文管理目前的 Agent 是“单次任务型”的。要支持多轮对话需要引入更复杂的记忆机制。对话历史 将整个messages列表持久化并在每次调用时作为上下文传入。注意上下文长度限制可能需要总结或裁剪历史。向量记忆 使用向量数据库如 Chroma, Pinecone存储历史交互的关键信息实现长期记忆和相似性检索。知识库 为 Agent 接入公司内部文档、技术手册等私有知识源使其回答更具专业性。7.2 实现更复杂的多 Agent 协作对于更复杂的任务可以引入多个特化的 Agent 协同工作。例如研究员 Agent 专门负责搜索和整理信息。分析师 Agent 负责从信息中提炼观点和趋势。写手 Agent 负责根据分析结果生成结构化的内容。 在 LangGraph 中这可以通过创建多个子图Subgraph或为不同节点绑定不同的 LLM 和工具集来实现。7.3 加入评估与人工反馈环节完全自主的 Agent 可能产生不符合要求的输出。在生产流程中引入评估节点至关重要。自动评估 增加一个evaluate_node用另一个 LLM 或规则引擎检查生成结果的质量如相关性、完整性、格式。人工介入点 在工作流的关键决策点如规划确认、最终输出前设置“暂停”等待人工审核或修改后再继续。反馈学习 将人工修正的结果保存下来用于微调模型或优化提示词。7.4 监控、日志与可观测性生产系统必须可观测。你需要记录每次 LLM 调用的输入、输出、token 消耗和耗时。每次工具调用的参数和结果。工作流每一步的状态变化。最终输出结果和用户满意度如果有反馈机制。 这些日志应输出到集中式的日志系统如 ELK Stack并设置关键指标如成功率、平均响应时间、平均成本的仪表盘。构建一个真正有价值的垂直领域 Agent技术实现只是第一步。更重要的是深刻理解业务场景设计出符合用户心智模型的工作流并建立起持续迭代和优化的机制。从本文提供的这个可运行的最小原型出发你可以逐步加入上述高级特性最终打造出一个能够稳定、可靠解决实际问题的智能体系统。 基于LangGraph构建AI Agent：从核心原理到工程实践
在实际 AI 应用开发中我们经常听到“Agent”这个概念。它被描述为能够理解目标、使用工具、与环境交互并完成复杂任务的智能体。从年初的狂热到年中的遍地开花再到如今冷静的讨论垂直领域 Agent 的开发似乎正面临一个拐点。很多开发者发现将一个概念验证PoC级别的 Agent 真正落地到生产环境解决实际的业务问题远比想象中要困难。这背后涉及的不只是模型 API 的调用更是一整套关于架构设计、工具集成、状态管理、异常处理和成本控制的工程实践。本文旨在为那些希望深入理解并动手构建 AI Agent 的开发者提供一个清晰的工程化视角。我们将暂时搁置对市场趋势的预测聚焦于一个核心问题如何从零开始搭建一个具备基本规划、执行和反思能力的 Agent 系统。我们将以当前社区中较为流行的 LangGraph 框架为例因为它清晰地体现了 Agent 的“有状态工作流”这一核心思想。通过构建一个模拟的“技术博客选题分析 Agent”你将理解 Agent 的核心组件、工作流程并掌握将其部署运行的关键步骤。最终你会获得一个可运行、可扩展的代码基础并了解在将其推向生产环境前必须考虑的工程化问题。1. 理解 Agent 的核心从概念到架构在开始写代码之前我们必须厘清几个关键概念。很多人将“调用大模型 API”等同于构建 Agent这是一个常见的误解。Agent 的本质是一个自主的决策系统。1.1 Agent 是什么不仅仅是提示词工程一个 AI Agent 通常由以下几个核心部分组成规划Planning Agent 需要理解复杂目标并将其分解为一系列可执行的子任务或步骤。这超越了简单的单轮问答。工具使用Tool Use Agent 能够调用外部工具如搜索引擎、计算器、数据库、API来获取信息或执行操作从而突破大模型自身知识的局限。记忆Memory Agent 需要具备短期记忆当前会话的上下文和长期记忆存储历史交互、知识以保持对话的一致性和进行学习。行动Action 基于规划、工具和记忆Agent 执行具体的操作可能是生成一段文本也可能是调用一个工具。反思Reflection 高级的 Agent 能够评估自身行动的结果判断任务是否完成或是否需要调整策略。这构成了一个“观察-思考-行动”的循环。因此构建 Agent 是一个系统工程它涉及提示词设计、工作流编排、状态管理、工具集成和异常处理等多个层面。1.2 为什么需要 Agent 框架LangGraph 的角色当你试图用代码实现上述循环时很快就会遇到挑战如何管理复杂的执行状态如何处理分支和循环如何让多个 Agent 协作手动用if-else和循环来控制流程会变得极其臃肿且难以维护。这就是 Agent 框架的价值所在。LangGraph是 LangChain 生态系统中的一个库它允许你将 Agent 系统建模为一个有向图Graph。图中的节点Node代表一个执行步骤如调用 LLM、运行工具边Edge代表步骤之间的流转条件。这种抽象完美地契合了 Agent 的“规划-执行-评估”循环。使用 LangGraph你可以清晰定义工作流 将 Agent 的逻辑可视化为一幅图便于理解和调试。管理复杂状态 框架帮你维护一个共享的状态字典在不同节点间传递和更新信息。处理循环和条件分支 轻松实现“直到任务完成才停止”或“根据结果选择不同路径”的逻辑。支持多 Agent 协作 可以将不同的节点分配给不同特化的 Agent如一个负责搜索一个负责写作。接下来我们将通过一个具体案例展示如何使用 LangGraph 构建一个 Agent。2. 环境准备与项目初始化我们将构建一个“技术博客选题分析 Agent”。它的目标是给定一个宽泛的技术领域如“云原生”Agent 能自动分析当前趋势生成几个具体的博客选题建议并为每个选题撰写一个简短的提纲。2.1 技术栈与依赖选择这个项目主要依赖 Python 和以下库LangChain / LangGraph: 用于构建 Agent 工作流。OpenAI API(或其它兼容 API): 为大模型提供能力。我们将使用gpt-4o-mini或gpt-4-turbo作为“大脑”。Tavily Search API: 一个为 AI 优化的搜索引擎工具用于获取实时网络信息。你也可以用 SerperAPI 或自己封装 Google Search。环境管理工具: 推荐使用conda或venv。以下是具体的依赖版本requirements.txtlangchain0.2.0 langchain-openai0.1.0 langchain-community0.0.10 langgraph0.0.52 tavily-python0.3.0 python-dotenv1.0.0注意LangChain 生态版本迭代较快以上版本为撰写时的稳定版本。实际安装前建议查看官方文档确认最新兼容版本。2.2 项目结构与关键文件创建一个清晰的项目目录结构是良好工程实践的开始。tech_blog_agent/ ├── .env # 存储API密钥等敏感信息 ├── requirements.txt # 项目依赖 ├── main.py # Agent工作流主入口 ├── agents/ │ ├── __init__.py │ ├── blog_agent.py # 定义Agent状态图和节点 │ └── tools.py # 定义自定义工具如搜索 ├── config/ │ └── settings.py # 配置管理模型、API端点等 └── run_agent.py # 运行脚本首先设置环境变量。在项目根目录创建.env文件# .env OPENAI_API_KEYsk-your-openai-api-key-here TAVILY_API_KEYyour-tavily-api-key-here重要 务必在.gitignore中加入.env避免将密钥提交到代码仓库。然后创建config/settings.py来集中管理配置# config/settings.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的环境变量 class Settings: OPENAI_API_KEY os.getenv(OPENAI_API_KEY) TAVILY_API_KEY os.getenv(TAVILY_API_KEY) # 模型配置 LLM_MODEL gpt-4o-mini # 也可用 gpt-4-turbo LLM_TEMPERATURE 0.1 # 较低的温度使输出更稳定 settings Settings()3. 构建核心 Agent 工作流我们将使用 LangGraph 的“状态图StateGraph”来定义 Agent。核心是定义一个共享的“状态”和一系列操作该状态的“节点”。3.1 定义 Agent 的状态State状态是一个字典包含了工作流执行过程中需要传递和更新的所有信息。我们为博客选题 Agent 设计如下状态# agents/blog_agent.py from typing import TypedDict, List, Annotated import operator from langgraph.graph.message import add_messages from langchain_core.messages import BaseMessage, HumanMessage class AgentState(TypedDict): # 输入用户最初的问题 input: str # 消息历史用于记录与LLM的对话 messages: Annotated[List[BaseMessage], add_messages] # 规划阶段产生的子任务列表 sub_tasks: List[str] # 搜索工具返回的原始资料 search_results: str # 生成的最终博客选题列表 blog_topics: List[dict] # 当前执行到了哪个步骤用于控制流程 current_step: strAnnotated[List[BaseMessage], add_messages]是 LangGraph 的语法糖它自动帮我们处理消息列表的追加非常方便。current_step字段将用于控制图的走向例如判断是否需要进行搜索。3.2 创建工具Tools工具是 Agent 的手臂。我们首先创建一个搜索工具用于获取最新的技术趋势信息。# agents/tools.py from langchain_community.tools.tavily_search import TavilySearchResults from config.settings import settings def get_search_tool(): 初始化并返回Tavily搜索工具 tool TavilySearchResults( api_keysettings.TAVILY_API_KEY, max_results3, # 每次搜索返回3条结果避免信息过载 search_depthadvanced, # 获取更详细的内容 ) return tool # 可以在这里定义更多工具比如获取GitHub趋势、查询技术文档等。3.3 定义工作流节点Nodes节点是图中的一个执行单元。我们将工作流分解为四个主要节点规划、研究、生成和路由。节点1规划节点plan_node这个节点接收用户输入让 LLM 分析并制定执行计划子任务。# agents/blog_agent.py from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from config.settings import settings llm ChatOpenAI(modelsettings.LLM_MODEL, temperaturesettings.LLM_TEMPERATURE) def plan_node(state: AgentState): 分析用户输入规划执行步骤 prompt ChatPromptTemplate.from_messages([ (system, 你是一个资深技术博客策划专家。你的任务是将用户宽泛的请求分解为具体的、可执行的研究步骤。 例如用户输入“云原生”你可能需要研究1. 最新的云原生技术趋势如服务网格、Serverless。2. 开发者常见的痛点。3. 社区热门讨论话题。 请只输出一个步骤列表每个步骤用一句话描述。), (human, 用户请求{input}), ]) chain prompt | llm # 调用LLM获取规划结果 plan_result chain.invoke({input: state[input]}) # 假设LLM返回的是文本我们按行解析成任务列表实际项目需要更鲁棒的解析 tasks [line.strip() for line in plan_result.content.split(\n) if line.strip()] # 更新状态 new_state { **state, sub_tasks: tasks, current_step: research, # 规划完成后进入研究阶段 messages: state[messages] [HumanMessage(contentf规划完成。子任务{tasks})], } return new_state节点2研究节点research_node此节点执行规划中的第一个研究性子任务例如搜索并保存结果。# agents/blog_agent.py from agents.tools import get_search_tool search_tool get_search_tool() def research_node(state: AgentState): 执行研究任务例如搜索 if not state[sub_tasks]: # 如果没有子任务直接跳过研究 return {**state, search_results: 无研究任务, current_step: generate} # 这里简化处理只取第一个子任务进行搜索 research_query state[sub_tasks][0] # 调用搜索工具 try: results search_tool.invoke(research_query) # Tavily 返回的是字典列表我们将其转换为易读的文本 result_text \n---\n.join([ f标题{r[title]}\n内容{r[content]}\n链接{r[url]} for r in results ]) except Exception as e: result_text f搜索工具调用失败{str(e)} new_state { **state, search_results: result_text, current_step: generate, # 研究完成后进入生成阶段 messages: state[messages] [HumanMessage(contentf研究完成。查询{research_query})], } return new_state节点3生成节点generate_node基于用户输入和研究结果生成具体的博客选题和提纲。# agents/blog_agent.py def generate_node(state: AgentState): 生成博客选题和提纲 prompt ChatPromptTemplate.from_messages([ (system, 你是一位经验丰富的技术作者。基于用户的需求和提供的研究资料生成3个具体、有价值、可落地的技术博客选题。 对于每个选题请提供 1. 标题吸引人且包含核心关键词。 2. 目标读者例如初级开发者、架构师、运维工程师。 3. 核心要点3-5个文章主要部分或论点。 4. 难点与亮点这篇文章写作的挑战和独特价值在哪里。 请以清晰的JSON格式输出包含一个“topics”数组每个元素是一个包含上述字段的对象。), (human, 用户原始需求{input} 研究资料{research} 请基于以上信息生成博客选题。 ), ]) chain prompt | llm response chain.invoke({ input: state[input], research: state.get(search_results, 无可用资料) }) # 这里需要解析LLM返回的JSON。实际项目中应使用LLM的JSON模式或后处理解析。 # 为简化我们假设LLM返回了格式良好的JSON字符串并用eval演示生产环境应用json.loads并加异常处理。 import json try: # 尝试从响应文本中提取JSON部分 import re json_match re.search(rjson\n(.*?)\n, response.content, re.DOTALL) if json_match: json_str json_match.group(1) else: json_str response.content topics_data json.loads(json_str) blog_topics topics_data.get(topics, []) except (json.JSONDecodeError, AttributeError) as e: blog_topics [{error: f解析LLM输出失败{str(e)}, raw_output: response.content[:200]}] new_state { **state, blog_topics: blog_topics, current_step: end, # 生成完成后工作流结束 messages: state[messages] [HumanMessage(content选题生成完成。)], } return new_state节点4路由节点router_node这是一个条件判断节点根据当前状态决定下一步走向哪个节点。# agents/blog_agent.py def router_node(state: AgentState) - str: 根据当前步骤决定下一个节点 current state.get(current_step, plan) if current plan: return plan elif current research: return research elif current generate: return generate else: return __end__ # LangGraph 内置的特殊节点表示结束3.4 组装成图Graph将节点和边组装起来形成一个完整的工作流。# agents/blog_agent.py from langgraph.graph import StateGraph, END def create_blog_agent_workflow(): 创建并返回博客选题Agent的工作流图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(plan, plan_node) workflow.add_node(research, research_node) workflow.add_node(generate, generate_node) # 设置入口点 workflow.set_entry_point(plan) # 添加条件边。router_node 函数决定下一个节点是谁。 workflow.add_conditional_edges( plan, router_node, { plan: plan, # 理论上不会发生仅为演示 research: research, generate: generate, __end__: END, } ) workflow.add_conditional_edges( research, router_node, { research: research, generate: generate, __end__: END, } ) workflow.add_conditional_edges( generate, router_node, { generate: generate, __end__: END, } ) # 编译图 app workflow.compile() return app4. 运行与验证现在我们已经有了一个完整的 Agent 工作流。接下来创建一个主程序来运行它并查看结果。4.1 创建运行脚本# run_agent.py from agents.blog_agent import create_blog_agent_workflow, AgentState from langchain_core.messages import HumanMessage import asyncio async def main(): # 1. 初始化工作流 app create_blog_agent_workflow() # 2. 准备初始状态 user_input 云原生监控 # 可以替换为任何技术主题如“大模型微调”、“React性能优化” initial_state: AgentState { input: user_input, messages: [HumanMessage(contentuser_input)], sub_tasks: [], search_results: , blog_topics: [], current_step: plan, } # 3. 运行工作流 print(f开始执行Agent分析主题{user_input}...) print(- * 50) final_state None # 我们可以流式输出每个步骤的结果便于观察 async for event in app.astream(initial_state, stream_modevalues): node_name list(event.keys())[0] state event[node_name] print(f[节点{node_name}] 当前步骤{state.get(current_step)}) if node_name plan and state.get(sub_tasks): print(f 规划的子任务{state[sub_tasks]}) elif node_name research and state.get(search_results): # 搜索结果可能很长只打印前500字符 preview (state[search_results][:500] ...) if len(state[search_results]) 500 else state[search_results] print(f 研究结果预览{preview}) elif node_name generate and state.get(blog_topics): print(f 生成的博客选题) for i, topic in enumerate(state[blog_topics], 1): print(f 选题{i}: {topic.get(title, N/A)}) print(- * 30) final_state state # 4. 输出最终结果 print(\n *50) print(最终结果) if final_state and final_state.get(blog_topics): for i, topic in enumerate(final_state[blog_topics], 1): print(f\n选题 {i}:) print(f 标题{topic.get(title)}) print(f 目标读者{topic.get(target_audience)}) print(f 核心要点{topic.get(key_points)}) print(f 难点与亮点{topic.get(challenges_and_highlights)}) else: print(未生成有效选题。) print(f最终状态{final_state}) if __name__ __main__: asyncio.run(main())4.2 执行与输出分析在终端运行python run_agent.py。你应该能看到类似以下的输出具体内容因模型和搜索实时结果而异开始执行Agent分析主题云原生监控... -------------------------------------------------- [节点plan] 当前步骤research 规划的子任务[1. 调研当前云原生监控领域的主流工具和框架如Prometheus, Grafana, Jaeger等。, 2. 分析在微服务、容器化环境下监控面临的独特挑战如动态性、多维度指标。, 3. 收集开发者/运维在云原生监控实践中的常见痛点与最佳实践。] ------------------------------ [节点research] 当前步骤generate 研究结果预览标题2024年云原生监控全景图从Prometheus到OpenTelemetry 内容随着Kubernetes和微服务的普及监控体系正从传统基础设施监控转向可观测性...后略 链接https://example.com/article1 --- 标题微服务监控的五大挑战与解决方案 内容动态扩缩容、服务间调用链追踪、日志聚合...后略 链接https://example.com/article2 ... ------------------------------ [节点generate] 当前步骤end 生成的博客选题 选题1: 《云原生可观测性入门搭建你的第一个PrometheusGrafana监控栈》 选题2: 《告别“监控孤岛”基于OpenTelemetry实现统一链路追踪》 选题3: 《Kubernetes集群监控实战核心指标解读与告警配置优化》 最终结果 选题 1: 标题《云原生可观测性入门搭建你的第一个PrometheusGrafana监控栈》 目标读者刚接触云原生的开发者和运维人员 核心要点1. 云原生监控与传统监控的区别。 2. Prometheus数据模型与核心组件。 3. 使用Helm在K8s中快速部署。 4. Grafana可视化仪表盘配置。 5. 编写第一条PromQL查询与告警规则。 难点与亮点难点在于理解Prometheus的拉取模型和指标类型亮点是提供从零到一的完整可运行示例。 选题 2: 标题《告别“监控孤岛”基于OpenTelemetry实现统一链路追踪》 目标读者中高级后端架构师和SRE 核心要点1. 分布式追踪的核心概念Trace, Span。 2. OpenTelemetry架构与数据收集流程。 3. 在Go/Java服务中集成OTel SDK。 4. 将追踪数据导出到Jaeger/Tempo。 5. 利用追踪数据进行性能瓶颈分析。 难点与亮点难点是多语言SDK的配置差异和数据关联亮点是打通Metrics、Logs、Traces的实践方案。 选题 3: 略这个输出表明你的 Agent 已经成功运行。它完成了“规划 - 研究 - 生成”的完整循环并输出了结构化的结果。5. 关键配置、参数与工程化考量一个能跑的 Demo 和一個健壯的生产级 Agent 之间存在巨大差距。以下是几个关键的工程化考量点。5.1 模型与参数配置模型的选择和参数调优直接影响 Agent 的稳定性、成本和质量。参数推荐值说明影响模型选择gpt-4o-mini/gpt-4-turbogpt-4o-mini成本低、速度快适合简单任务和频繁调用。gpt-4-turbo推理能力更强适合复杂规划和分析。成本、响应速度、任务完成质量。Temperature0.1 - 0.3控制输出的随机性。对于需要稳定、结构化输出的 Agent 任务应设置较低的值。值越高输出越有创造性但可能不稳定值越低输出越确定但可能死板。Max Tokens根据任务设定限制单次响应的最大长度。对于规划、总结等任务可设小些如500对于生成长文需设大如2000。防止响应过长控制单次调用成本。Top P0.9 - 1.0与 Temperature 类似控制采样范围。通常与 Temperature 配合使用保持默认即可。影响输出的多样性和可预测性。在config/settings.py中集中管理这些参数# config/settings.py (扩展) class Settings: # ... 其他配置 ... LLM_MODEL gpt-4o-mini LLM_TEMPERATURE 0.1 LLM_MAX_TOKENS 1000 LLM_TOP_P 1.0 # 流式输出开关用于需要实时展示的场景 LLM_STREAMING False5.2 工具调用的错误处理与降级工具调用如网络搜索是 Agent 失败的主要风险点。必须实现健壮的错误处理。# agents/tools.py (增强版) import asyncio from typing import Optional from langchain_core.tools import ToolException def safe_tool_call(tool, input_str, max_retries2, timeout10): 带重试和超时的安全工具调用 for attempt in range(max_retries): try: # 使用异步或线程池防止阻塞这里简化用同步调用加超时 # 实际项目中如果框架支持应使用异步工具调用。 result tool.invoke(input_str) return result except ToolException as e: print(f工具调用失败 (尝试 {attempt1}/{max_retries}): {e}) if attempt max_retries - 1: raise asyncio.sleep(1) # 简单等待后重试 except Exception as e: # 网络超时、API限制等通用异常 print(f工具调用发生未知错误: {e}) return f工具暂时不可用{str(e)} return 工具调用失败请稍后重试。 # 在research_node中调用 # result_text safe_tool_call(search_tool, research_query)5.3 状态管理与持久化LangGraph 的内存状态在进程结束后就消失了。对于需要长期对话或异步执行的任务必须将状态持久化。# 示例使用数据库如SQLite持久化状态 import sqlite3 import json from datetime import datetime class StateManager: def __init__(self, db_pathagent_state.db): self.conn sqlite3.connect(db_path) self._init_db() def _init_db(self): cursor self.conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS workflow_state ( session_id TEXT PRIMARY KEY, state_json TEXT NOT NULL, updated_at TIMESTAMP NOT NULL ) ) self.conn.commit() def save_state(self, session_id: str, state: dict): cursor self.conn.cursor() state_json json.dumps(state) cursor.execute( INSERT OR REPLACE INTO workflow_state (session_id, state_json, updated_at) VALUES (?, ?, ?) , (session_id, state_json, datetime.utcnow())) self.conn.commit() def load_state(self, session_id: str) - Optional[dict]: cursor self.conn.cursor() cursor.execute(SELECT state_json FROM workflow_state WHERE session_id?, (session_id,)) row cursor.fetchone() if row: return json.loads(row[0]) return None # 在运行工作流时可以加载和保存状态 # state_manager StateManager() # saved_state state_manager.load_state(user_session_id) # if saved_state: # initial_state.update(saved_state) # final_state app.invoke(initial_state) # state_manager.save_state(user_session_id, final_state)6. 常见问题排查与调试在开发 Agent 过程中你会遇到各种问题。以下是一个快速排查清单。6.1 Agent 流程卡住或陷入循环现象可能原因检查与解决工作流一直在plan和research节点间循环。router_node逻辑错误或current_step状态更新不正确。1. 在router_node和每个节点函数内打印state[‘current_step’]。2. 确保每个节点执行后都正确更新了current_step。3. 检查条件边的映射关系是否正确。LLM 输出不符合预期导致后续节点解析失败。提示词Prompt不够清晰或 Temperature 设置过高。1. 在 Prompt 中明确指定输出格式如“请输出 JSON”。2. 使用 LLM 的 JSON 模式如response_format{“type”: “json_object”}。3. 将 Temperature 调低至 0.1。4. 在代码中添加对 LLM 输出的后处理如正则提取、重试。工具调用超时或返回错误。网络问题、API 密钥无效、工具参数错误。1. 为工具调用添加超时和重试机制见 5.2 节。2. 检查环境变量是否正确加载。3. 单独测试工具调用是否正常print(search_tool.invoke(“test”))。6.2 输出质量低下现象可能原因检查与解决生成的选题过于宽泛或空洞。1. 用户输入太宽泛。2. 搜索工具返回的结果质量差。3. 生成节点的 Prompt 约束力不够。1. 引导用户输入更具体的问题或在规划节点让 LLM 先澄清需求。2. 优化搜索查询词或更换/组合多个搜索工具。3. 在生成 Prompt 中提供更具体的范例Few-shot Prompting。选题之间重复度高。LLM 在生成多个条目时缺乏多样性。1. 在 Prompt 中明确要求“角度不同”、“覆盖不同子领域”。2. 让生成节点分两次或多次调用 LLM每次聚焦一个子方向。3. 在生成后添加一个“去重/排序”节点。6.3 性能与成本问题现象可能原因检查与解决单次请求耗时过长30秒。1. 串行调用工具和 LLM。2. 网络延迟高。3. 使用了速度慢的大模型如 GPT-4。1. 分析工作流将可以并行的节点并行化LangGraph 支持。2. 为网络请求设置合理超时并使用更稳定的工具提供商。3. 在非关键路径使用更快、更便宜的模型如用gpt-4o-mini做规划用gpt-4-turbo做最终生成。API 调用费用飙升。1. 工作流设计复杂调用链路过长。2. 每次调用都使用高 token 数的模型。3. 没有缓存重复内容。1. 优化工作流减少不必要的 LLM 调用例如缓存规划结果。2. 在config/settings.py中为不同节点配置不同的模型。3. 对搜索等工具的结果进行去重和缓存避免对相同查询重复调用。7. 从 Demo 到生产最佳实践与扩展方向要让这个 Agent 真正可用还需要在以下几个方面进行强化。7.1 增强记忆与上下文管理目前的 Agent 是“单次任务型”的。要支持多轮对话需要引入更复杂的记忆机制。对话历史 将整个messages列表持久化并在每次调用时作为上下文传入。注意上下文长度限制可能需要总结或裁剪历史。向量记忆 使用向量数据库如 Chroma, Pinecone存储历史交互的关键信息实现长期记忆和相似性检索。知识库 为 Agent 接入公司内部文档、技术手册等私有知识源使其回答更具专业性。7.2 实现更复杂的多 Agent 协作对于更复杂的任务可以引入多个特化的 Agent 协同工作。例如研究员 Agent 专门负责搜索和整理信息。分析师 Agent 负责从信息中提炼观点和趋势。写手 Agent 负责根据分析结果生成结构化的内容。 在 LangGraph 中这可以通过创建多个子图Subgraph或为不同节点绑定不同的 LLM 和工具集来实现。7.3 加入评估与人工反馈环节完全自主的 Agent 可能产生不符合要求的输出。在生产流程中引入评估节点至关重要。自动评估 增加一个evaluate_node用另一个 LLM 或规则引擎检查生成结果的质量如相关性、完整性、格式。人工介入点 在工作流的关键决策点如规划确认、最终输出前设置“暂停”等待人工审核或修改后再继续。反馈学习 将人工修正的结果保存下来用于微调模型或优化提示词。7.4 监控、日志与可观测性生产系统必须可观测。你需要记录每次 LLM 调用的输入、输出、token 消耗和耗时。每次工具调用的参数和结果。工作流每一步的状态变化。最终输出结果和用户满意度如果有反馈机制。 这些日志应输出到集中式的日志系统如 ELK Stack并设置关键指标如成功率、平均响应时间、平均成本的仪表盘。构建一个真正有价值的垂直领域 Agent技术实现只是第一步。更重要的是深刻理解业务场景设计出符合用户心智模型的工作流并建立起持续迭代和优化的机制。从本文提供的这个可运行的最小原型出发你可以逐步加入上述高级特性最终打造出一个能够稳定、可靠解决实际问题的智能体系统。