尧图网页设计 
如果你正在开发基于大语言模型LLM的应用是否遇到过这样的困境模型回答看似流畅但一到具体业务场景就“胡说八道”智能体Agent设计时逻辑链一长就失控要么陷入死循环要么忘记核心任务或者团队协作时每个人写的提示词Prompt风格迥异导致效果天差地别难以维护这些问题本质上都不是模型能力的问题而是工程化的问题。大语言模型本身只是一个强大的“大脑”而如何为这个大脑构建稳定、可靠、可协作的“神经系统”和“行动躯体”才是决定AI应用成败的关键。这正是“大语言模型工程”LLM Engineering要解决的核心命题。本文是“大语言模型工程”系列的第二部分。在第一部分我们探讨了基础概念与RAG检索增强生成之后本篇将深入更复杂的领域智能体Agent的开发、工程化实践与前沿架构。我们将不再停留在“调用API”的层面而是聚焦于如何设计一个能自主规划、使用工具、并可靠执行复杂任务的智能系统。你会看到从简单的函数调用到基于LangGraph的工作流编排从提示工程到智能体的“人设”与记忆管理每一步都充满了工程上的权衡与设计。读完本文你将能清晰地构建一个可落地的智能体开发框架理解当前主流工具链如LangChain, LangGraph, MCP的最佳实践并避开智能体开发中最常见的那些“坑”。1. 智能体开发从“玩具”到“工具”的关键跨越很多人对智能体Agent的第一印象可能来自AutoGPT或ChatGPT的插件功能——能自动上网搜索、写代码、分析数据。然而将这些演示转化为真正解决业务问题、稳定运行的生产级应用中间隔着一道巨大的工程鸿沟。一个“玩具级”智能体和“工具级”智能体的核心区别在于可靠性、可控性和可观测性。可靠性能否在百次、千次调用中稳定完成目标而非偶尔成功可控性能否在任务偏离轨道时被干预或纠正能否理解其决策过程可观测性开发者和用户能否清晰地知道智能体“在想什么”、“做了什么”、“为什么失败”例如一个简单的“总结网页内容”任务玩具级智能体可能直接调用搜索工具并生成总结。但工具级智能体需要考虑如果搜索不到结果怎么办如果网页内容过长超出上下文窗口怎么办如果总结出现事实性幻觉Hallucination如何校验这些边界情况的处理正是LLM工程的核心。因此智能体开发远不止是写一段聪明的提示词。它是一个系统工程涉及规划Planning、工具使用Tool Use、记忆Memory、执行Execution和评估Evaluation等多个模块的协同。接下来我们将从基础概念开始逐步拆解如何搭建这样一个系统。2. 核心概念辨析Agent、Tool、Memory与Workflow在深入实践前必须厘清几个容易混淆的核心概念。它们构成了智能体系统的骨架。2.1 智能体Agent vs. 链Chain链Chain是预定义的、线性的执行流程。例如“检索文档 - 生成摘要”是一个链。它的路径是固定的适合确定性高的任务。智能体Agent则具备动态决策能力。它根据当前状态和目标自主决定下一步调用哪个工具甚至是否需要进行多步规划。它适合开放域、路径不确定的复杂任务。简单说链是“铁轨”智能体是“驾驶员”。在LangChain等框架中Agent本质上是一个特殊的Chain它内部封装了决策逻辑通常由一个大语言模型驱动。2.2 工具Tool工具是智能体与外部世界交互的“手”和“脚”。一个工具通常对应一个具体的函数例如search_web(query): 执行网络搜索。execute_sql(sql_query): 查询数据库。call_api(endpoint, params): 调用内部API。read_file(file_path): 读取本地文件。智能体的强大之处在于它能动态组合使用这些工具来完成复杂目标。工程上的关键点在于如何清晰、安全地向LLM描述工具的功能和输入格式。2.3 记忆Memory记忆决定了智能体的“上下文”有多长以及它如何利用历史信息。主要分为两类短期记忆Short-term Memory即对话上下文窗口。通常通过维护一个对话历史列表来实现。长期记忆Long-term Memory超越单次对话周期的信息存储。例如将重要的对话摘要或用户偏好存入向量数据库供后续会话检索。这对于构建有“人设”或持续学习的智能体至关重要。2.4 工作流Workflow与编排Orchestration当单个智能体无法处理过于复杂的任务时我们需要引入工作流。工作流将一个大任务分解为多个子任务并由不同的智能体或节点Node协作完成。编排则是管理这些节点执行顺序和状态流转的机制。近年来LangGraph基于LangChain和MCPModel Context Protocol等框架的兴起正是为了更优雅地解决复杂工作流的编排问题。它们允许你以图Graph的形式定义状态机清晰描述“如果条件A成立则执行B否则执行C”这样的逻辑。3. 环境准备构建你的智能体开发沙箱在开始编码前我们需要搭建一个稳定且高效的开发环境。以下配置是一个兼顾灵活性和生产实践的起点。核心工具栈选择Python 3.10LLM生态的主流语言。LangChain LangGraph当前智能体开发的事实标准框架社区活跃工具链丰富。OpenAI API 或 本地大模型推理核心。为快速原型开发建议使用GPT-4系列API为追求成本可控和数据隐私可部署本地模型如Qwen、DeepSeek或Llama 3。Jupyter Notebook / VS Code交互式开发和调试。向量数据库可选如需长期记忆或RAG能力可准备Chroma、Weaviate或PGVector。3.1 基础环境搭建首先创建并激活一个独立的Python虚拟环境这是管理依赖冲突的最佳实践。# 创建项目目录并进入 mkdir llm-agent-project cd llm-agent-project # 创建虚拟环境以venv为例 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate3.2 依赖安装创建requirements.txt文件包含以下核心依赖# 核心框架 langchain0.1.0 langchain-openai0.0.5 langgraph0.0.30 # 可选用于连接本地模型或其它API langchain-community0.0.10 # 工具与工具调用相关示例 langchain-experimental0.0.50 # 包含一些实验性但有用的Agent # 实用工具 python-dotenv1.0.0 # 管理环境变量 jupyter1.0.0 # 用于交互式探索 # 可选向量数据库客户端以Chroma为例 chromadb0.4.22使用pip进行安装pip install -r requirements.txt3.3 配置API密钥强烈建议使用环境变量管理敏感信息如API Key。创建.env文件并确保将其加入.gitignore# .env 文件示例 OPENAI_API_KEYsk-your-openai-api-key-here # 如果使用其他模型例如通义千问 DASHSCOPE_API_KEYyour-dashscope-key # 如果使用向量数据库 CHROMA_PERSIST_DIRECTORY./chroma_db在代码中通过dotenv加载# config.py import os from dotenv import load_dotenv load_dotenv() OPENAI_API_KEY os.getenv(OPENAI_API_KEY) if not OPENAI_API_KEY: raise ValueError(请在 .env 文件中设置 OPENAI_API_KEY)至此你的智能体开发沙箱就准备就绪了。4. 从零构建你的第一个智能体一个任务规划助手让我们从一个具体的例子开始构建一个“周末活动规划助手”。这个智能体需要根据用户的简单需求如“我想在周末做些放松又有趣的事预算中等”自动调用工具来搜索信息、评估选项并生成一个可行的计划。4.1 第一步定义工具Tools工具是智能体的能力边界。我们定义三个简单的工具# tools.py from langchain.tools import tool from typing import Optional tool def search_local_events(city: str, date: str, category: Optional[str] None) - str: 根据城市、日期和可选类别搜索本地活动如展览、音乐会、市集。 这是一个模拟函数实际项目中应接入真实数据API。 # 模拟数据 events_db { 北京: { 2024-05-25: [故宫博物院特展, 798艺术区新潮市集, 国家大剧院音乐会], 2024-05-26: [奥林匹克公园骑行活动, 三里屯咖啡文化节] }, 上海: { 2024-05-25: [上海当代艺术博物馆展览, 迪士尼乐园特别活动], 2024-05-26: [外滩源周末市集, 浦东图书馆讲座] } } events events_db.get(city, {}).get(date, []) if category: # 简单模拟类别过滤 events [e for e in events if category in e] return f在{city}{date}找到的活动{, .join(events) if events else 暂无活动} tool def get_weather(city: str, date: str) - str: 获取指定城市和日期的天气预报。 # 模拟天气数据 weather_data { 北京: {2024-05-25: 晴15-25°C, 2024-05-26: 多云18-27°C}, 上海: {2024-05-25: 小雨20-28°C, 2024-05-26: 阴22-30°C} } forecast weather_data.get(city, {}).get(date, 暂无预报信息) return f{city}{date}的天气{forecast} tool def estimate_budget(activity: str, people: int 2) - str: 估算某项活动的大致人均预算单位元。 # 简单的预算估算逻辑 budget_map { 展览: 100, 音乐会: 300, 市集: 50, 骑行: 20, 咖啡: 80, 讲座: 0, 迪士尼: 500 } base_cost 0 for key, cost in budget_map.items(): if key in activity: base_cost cost break total base_cost * people return f活动「{activity}」{people}人参与预估总花费约{total}元人均{base_cost}元。4.2 第二步创建智能体Agent我们将使用LangChain的create_react_agent这是一种基于“Reasoning Acting”模式的强大智能体。# agent_basic.py from langchain import hub from langchain.agents import create_react_agent, AgentExecutor from langchain_openai import ChatOpenAI from tools import search_local_events, get_weather, estimate_budget from config import OPENAI_API_KEY # 1. 初始化大语言模型 llm ChatOpenAI( modelgpt-4-turbo-preview, # 对于复杂规划任务建议使用能力更强的模型 temperature0.2, # 较低的温度使输出更稳定、更可预测 api_keyOPENAI_API_KEY ) # 2. 准备工具列表 tools [search_local_events, get_weather, estimate_budget] # 3. 从LangChain Hub获取一个优化的ReAct提示词模板 # 提示词工程是Agent性能的关键使用社区验证过的模板是很好的起点。 prompt hub.pull(hwchase17/react) # 4. 创建ReAct Agent agent create_react_agent(llm, tools, prompt) # 5. 创建Agent执行器它负责运行Agent并处理工具调用 agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, # 开启详细日志便于调试和理解Agent的思考过程 handle_parsing_errorsTrue, # 优雅处理模型输出解析错误 max_iterations5, # 限制最大迭代次数防止无限循环 early_stopping_methodgenerate # 当Agent认为任务完成时可以提前停止 )4.3 第三步运行与测试现在让我们用这个智能体来回答一个用户查询。# 运行智能体 if __name__ __main__: query 我这个周末2024-05-25在北京想找一些预算在每人200元以内的户外活动请帮我规划一下并考虑天气因素。 print(f用户提问: {query}\n) print(*50) try: result agent_executor.invoke({input: query}) print(\n *50) print(最终回答:) print(result[output]) except Exception as e: print(f执行过程中出现错误: {e})预期输出与过程解析verboseTrue时用户提问: 我这个周末2024-05-25在北京想找一些预算在每人200元以内的户外活动请帮我规划一下并考虑天气因素。 进入新的Agent执行链... 思考用户在北京日期是2024-05-25需要户外活动人均预算200以内。我需要先查天气再找活动最后估算预算。 行动get_weather 行动输入{city: 北京, date: 2024-05-25} 观察北京2024-05-25的天气晴15-25°C 思考天气很好适合户外。现在搜索北京当天的活动。 行动search_local_events 行动输入{city: 北京, date: 2024-05-25, category: 户外} 观察在北京2024-05-25找到的活动奥林匹克公园骑行活动 思考找到了一个户外活动。现在需要估算其预算是否满足要求。 行动estimate_budget 行动输入{activity: 奥林匹克公园骑行活动, people: 1} 观察活动「奥林匹克公园骑行活动」1人参与预估总花费约20元人均20元。 思考人均20元远低于200元预算符合要求。我可以基于这些信息给出建议了。 最终回答: 根据您的要求我为您规划了2024-05-25周六在北京的活动 1. **活动建议**奥林匹克公园骑行活动。这是一个户外活动非常适合晴朗的天气当天预报晴15-25°C。 2. **预算分析**该活动人均花费约20元远低于您设定的200元预算。 3. **综合建议**这是一个性价比很高的选择。建议您准备好骑行装备并做好防晒。由于是户外活动请根据实时天气情况调整着装。通过verboseTrue的日志你可以清晰地看到智能体的“思考-行动-观察”循环ReAct Loop这是理解和调试Agent行为的最重要窗口。5. 进阶使用LangGraph构建可控的复杂工作流基础的Agent能处理简单任务但对于需要严格步骤、条件分支或多人协作的复杂流程我们就需要更强大的编排能力。LangGraph允许我们将工作流定义为一个有向图Graph其中节点是函数或子智能体边定义了执行路径。让我们构建一个更复杂的“智能旅行规划师”它包含决策分支和状态管理。5.1 定义状态State状态是贯穿整个工作流的数据容器。# travel_planner_graph.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END # 1. 定义状态结构 class TravelPlanState(TypedDict): 旅行规划的工作流状态 user_request: str # 原始用户请求 destination: str # 解析出的目的地 dates: str # 解析出的日期 traveler_count: int # 旅行人数 budget_constraint: str # 预算约束 # 中间结果 weather_info: str flight_options: List[str] hotel_options: List[str] activity_options: List[str] # 最终输出 final_itinerary: str warnings: List[str]5.2 创建节点Nodes每个节点是一个处理特定任务的函数。# 2. 定义各个节点函数 def parse_user_request(state: TravelPlanState) - TravelPlanState: 节点1解析用户请求提取关键信息 from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0) prompt ChatPromptTemplate.from_messages([ (system, 你是一个旅行请求解析助手。从用户请求中提取目的地、日期、人数和预算信息。), (human, {request}) ]) chain prompt | llm # 为简化示例这里直接模拟解析结果 # 实际应用中应使用Pydantic或类似方法进行结构化输出 state[destination] 上海 state[dates] 2024-06-01 至 2024-06-03 state[traveler_count] 2 state[budget_constraint] 中等预算 print(f[节点-解析] 已解析请求: 目的地{state[destination]}, 日期{state[dates]}) return state def check_weather_and_warn(state: TravelPlanState) - TravelPlanState: 节点2检查目的地天气并生成警告 # 模拟天气查询 state[weather_info] 上海2024-06-01至2024-06-03多云转小雨气温22-30°C。 if 小雨 in state[weather_info]: state.setdefault(warnings, []).append(出行期间可能有雨建议携带雨具。) print(f[节点-天气] 天气信息已获取: {state[weather_info]}) return state def search_flights(state: TravelPlanState) - TravelPlanState: 节点3搜索航班模拟 # 模拟航班搜索实际应调用API state[flight_options] [ 航班A: 北京-上海 08:00-10:15 经济舱 ¥1200, 航班B: 北京-上海 14:30-16:45 经济舱 ¥950 ] print(f[节点-航班] 找到 {len(state[flight_options])} 个航班选项) return state def search_hotels(state: TravelPlanState) - TravelPlanState: 节点4搜索酒店模拟 state[hotel_options] [ 酒店X: 市中心 四星级 ¥600/晚 评分4.5, 酒店Y: 近景区 三星级 ¥400/晚 评分4.2 ] print(f[节点-酒店] 找到 {len(state[hotel_options])} 个酒店选项) return state def search_activities(state: TravelPlanState) - TravelPlanState: 节点5搜索当地活动模拟 state[activity_options] [ 外滩观光, 迪士尼乐园一日游, 豫园文化体验 ] print(f[节点-活动] 找到 {len(state[activity_options])} 个活动选项) return state def generate_itinerary(state: TravelPlanState) - TravelPlanState: 节点6整合所有信息生成最终行程计划 itinerary f 上海旅行计划{state[dates]} **出行人数:** {state[traveler_count]}人 **预算水平:** {state[budget_constraint]} **天气情况:** {state[weather_info]} **推荐航班:** {chr(10).join(f- {f} for f in state.get(flight_options, []))} **推荐酒店:** {chr(10).join(f- {h} for h in state.get(hotel_options, []))} **推荐活动:** {chr(10).join(f- {a} for a in state.get(activity_options, []))} **温馨提示:** {chr(10).join(f- {w} for w in state.get(warnings, [])) if state.get(warnings) else 无} state[final_itinerary] itinerary print([节点-生成] 行程计划已生成) return state5.3 构建图Graph并编排流程# 3. 构建工作流图 def build_travel_planner_graph(): # 初始化图 workflow StateGraph(TravelPlanState) # 添加节点 workflow.add_node(parse_request, parse_user_request) workflow.add_node(check_weather, check_weather_and_warn) workflow.add_node(search_flights, search_flights) workflow.add_node(search_hotels, search_hotels) workflow.add_node(search_activities, search_activities) workflow.add_node(generate_itinerary, generate_itinerary) # 定义边执行顺序 workflow.set_entry_point(parse_request) workflow.add_edge(parse_request, check_weather) # 并行执行航班、酒店、活动搜索 workflow.add_edge(check_weather, search_flights) workflow.add_edge(check_weather, search_hotels) workflow.add_edge(check_weather, search_activities) # 等待所有并行节点完成后进入生成节点 workflow.add_edge(search_flights, generate_itinerary) workflow.add_edge(search_hotels, generate_itinerary) workflow.add_edge(search_activities, generate_itinerary) workflow.add_edge(generate_itinerary, END) # 编译图 return workflow.compile() # 4. 运行工作流 if __name__ __main__: # 初始化状态 initial_state: TravelPlanState { user_request: 我想六月初去上海玩三天两个人中等预算。, destination: , dates: , traveler_count: 0, budget_constraint: , weather_info: , flight_options: [], hotel_options: [], activity_options: [], final_itinerary: , warnings: [] } print(开始执行旅行规划工作流...\n) app build_travel_planner_graph() final_state app.invoke(initial_state) print(\n *60) print(工作流执行完成最终行程计划) print(*60) print(final_state[final_itinerary])这个例子展示了LangGraph的核心优势显式的工作流定义。你可以清晰地看到每个步骤节点和它们的依赖关系。对于需要严格SOP标准作业程序的业务流程如图中所示的并行搜索这种基于图的编排比传统线性Agent可靠得多。6. 工程化实践提升智能体的可靠性与可维护性将智能体从Demo推向生产必须解决可靠性、可维护性和团队协作问题。以下是几个关键的工程实践。6.1 结构化输出Structured Outputs让LLM返回结构化数据如JSON、Pydantic模型而非自由文本是后续程序化处理的基础。这能极大减少解析错误。# structured_agent.py from langchain_openai import ChatOpenAI from langchain_core.pydantic_v1 import BaseModel, Field from langchain_core.prompts import ChatPromptTemplate from typing import List # 1. 定义期望的输出结构 class Itinerary(BaseModel): 行程计划的数据模型 destination: str Field(description旅行目的地) start_date: str Field(description开始日期) end_date: str Field(description结束日期) recommended_flights: List[str] Field(description推荐的航班列表) total_estimated_cost: float Field(description预估总花费) notes: List[str] Field(description注意事项) # 2. 创建支持结构化输出的链 llm ChatOpenAI(modelgpt-4-turbo-preview) structured_llm llm.with_structured_output(Itinerary) # 关键方法 prompt ChatPromptTemplate.from_messages([ (system, 你是一个旅行规划专家。根据用户请求生成结构化的行程计划。), (human, 用户请求{request}) ]) structured_chain prompt | structured_llm # 3. 调用并获取结构化对象 result structured_chain.invoke({ request: 我想下个月去杭州玩两天预算5000元左右。 }) print(f目的地: {result.destination}) print(f日期: {result.start_date} 至 {result.end_date}) print(f预估花费: ¥{result.total_estimated_cost}) print(注意事项:) for note in result.notes: print(f - {note})6.2 智能体的人设Persona与系统提示词工程系统提示词System Prompt是智能体的“人设”和“行为准则”。一个好的提示词能显著提升输出质量和稳定性。# persona_prompt.py from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from tools import search_local_events, get_weather # 复用之前的工具 # 定义一个专业、细致、谨慎的“旅行顾问”人设 system_prompt 你是一位资深的旅行规划顾问名叫“智行”。你的性格特点是 1. **专业严谨**提供的所有信息如价格、时间、地址都必须力求准确不确定时要明确说明。 2. **考虑周全**主动考虑用户未提及的因素如天气、交通、季节性活动、安全提醒。 3. **预算敏感**始终将用户的预算约束放在首位提供不同价位的选择并对比优劣。 4. **主动确认**在做出关键建议如预订类项目前应询问用户是否需要更详细的信息或确认。 5. **诚实透明**如果某些信息无法获取或工具调用失败如实告知用户而不是编造。 请根据以上原则调用合适的工具来回答用户的问题。 prompt ChatPromptTemplate.from_messages([ (system, system_prompt), MessagesPlaceholder(variable_namechat_history, optionalTrue), # 支持对话历史 (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), # Agent思考过程占位符 ]) llm ChatOpenAI(modelgpt-4-turbo, temperature0.3) tools [search_local_events, get_weather] # 使用OpenAI最新的tools calling格式创建Agent agent create_openai_tools_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) # 测试 result agent_executor.invoke({ input: 我周末想去户外走走但听说可能会下雨有什么建议吗, chat_history: [] # 如果是连续对话这里传入历史消息 })6.3 记忆Memory管理为了让智能体在长对话中保持连贯需要有效管理记忆。LangChain提供了多种记忆方案。# memory_management.py from langchain.memory import ConversationBufferWindowMemory, CombinedMemory, ConversationSummaryMemory from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI # 组合多种记忆策略 memory CombinedMemory( memories[ ConversationBufferWindowMemory( memory_keychat_history, # 存储最近几轮对话原文 k5, # 保留最近5轮对话 input_keyinput, output_keyoutput ), ConversationSummaryMemory( memory_keysummary_memory, # 存储对话摘要用于长期上下文 llmChatOpenAI(modelgpt-3.5-turbo, temperature0), input_keyinput ) ] ) # 在创建Agent时传入memory llm ChatOpenAI(modelgpt-4-turbo) tools [...] # 你的工具列表 prompt ... # 包含chat_history和summary_memory占位符的提示词 agent create_openai_tools_agent(llm, tools, prompt) agent_executor AgentExecutor( agentagent, toolstools, memorymemory, verboseTrue ) # 进行多轮对话 agent_executor.invoke({input: 我喜欢看科幻电影。}) agent_executor.invoke({input: 根据我的喜好推荐一部最近上映的。}) # Agent能记住上一轮的喜好6.4 测试与评估Evaluation如何衡量一个智能体的好坏不能只靠人工看几次对话。需要建立自动化的评估体系。# agent_evaluation.py import asyncio from langchain.evaluation import load_evaluator from langchain.evaluation import EvaluatorType # 1. 准备测试用例 test_cases [ { input: 北京明天天气怎么样, expected_actions: [get_weather] # 期望调用的工具 }, { input: 帮我规划一个预算500元的北京一日游, expected_actions: [get_weather, search_local_events, estimate_budget], expected_keywords: [预算, 500, 推荐] # 期望回答中包含的关键词 } ] # 2. 使用LangChain内置评估器 evaluator load_evaluator(EvaluatorType.QA) # 也可以使用CriteriaEvalChain来评估是否符合特定标准如“友好性”、“相关性” async def evaluate_agent(agent_executor, test_case): 异步评估单个测试用例 try: result await agent_executor.ainvoke({input: test_case[input]}) actual_output result[output] # 简单评估检查是否调用了期望的工具可通过日志解析 # 以及输出是否包含关键词 score 0 for keyword in test_case.get(expected_keywords, []): if keyword in actual_output: score 1 # 更复杂的评估可以使用LLM作为裁判 # eval_result await evaluator.aevaluate_strings( # predictionactual_output, # inputtest_case[input], # referencetest_case.get(expected_output, ) # ) return { input: test_case[input], output: actual_output, score: score, passed: score len(test_case.get(expected_keywords, [])) * 0.8 # 80%关键词命中即通过 } except Exception as e: return {input: test_case[input], error: str(e), passed: False} # 3. 批量运行评估 async def run_evaluations(): tasks [evaluate_agent(agent_executor, tc) for tc in test_cases] results await asyncio.gather(*tasks) passed sum(1 for r in results if r.get(passed, False)) print(f评估完成。通过率: {passed}/{len(results)} ({passed/len(results)*100:.1f}%)) for r in results: print(f输入: {r[input][:50]}... | 通过: {r.get(passed, N/A)} | 得分: {r.get(score, N/A)}) # 注意需要在一个异步环境中运行 # asyncio.run(run_evaluations())7. 常见问题与排查指南在智能体开发过程中你几乎一定会遇到以下问题。这里提供快速的排查思路。问题现象可能原因排查方式解决方案Agent陷入循环反复调用同一工具或思考。1. 提示词未明确停止条件。2. 工具返回信息不足以让LLM做出决策。3.max_iterations设置过高。1. 查看verboseTrue日志观察思考过程。2. 检查工具返回内容是否清晰、格式是否标准。1. 在系统提示词中明确“当你拥有足够信息时请直接给出最终答案”。2. 优化工具返回格式增加明确的状态标识如“搜索完成共X条结果”。3. 合理设置max_iterations通常5-10次。工具调用参数错误如类型不匹配或缺少参数。1. 工具函数描述docstring不清晰。2. LLM未能正确理解工具输入格式。1. 检查工具函数的类型注解和docstring是否准确描述了每个参数。2. 使用agent_executor.invoke并捕获异常查看具体错误。1. 使用Pydantic模型严格定义工具输入。2. 在提示词中举例说明工具调用格式。3. 使用OpenAI的tools格式而非旧的bind_tools其参数解析更准确。LLM不调用工具直接基于自身知识回答。1. 问题太简单LLM认为自己能直接回答。2. 工具描述不够有吸引力或必要。1. 询问一个必须调用工具才能回答的问题如“今天上海天气如何”。2. 检查工具描述是否强调了其独特价值如“获取实时数据”。1. 在系统提示词中强调“你必须使用工具来获取最新或特定信息”。2. 对简单问题可以接受LLM直接回答对需要准确性的问题强制其调用工具。响应速度慢。1. 工具本身是慢速IO操作如网络请求。2. LLM生成速度慢大模型或网络延迟。3. Agent迭代次数过多。1. 使用日志记录每个工具调用耗时。2. 检查网络连接和API响应时间。1. 为慢速工具设置超时timeout和重试机制。2. 考虑使用更快的LLM如GPT-3.5-Turbo进行简单决策。3. 使用异步调用ainvoke并行执行独立工具。生产环境部署后不稳定。1. API密钥、网络等环境问题。2. 提示词或工具版本变更。3. 流量突增导致限流。1. 实现完善的日志和监控记录每次调用的输入、输出、耗时和错误。2. 进行版本化管理每次变更前在测试环境充分验证。1. 增加健壮的错误处理retry, fallback。2. 使用配置中心管理提示词和参数。3. 实现限流和熔断机制保护下游服务。8. 最佳实践与架构建议基于大量的项目经验以下是一些能让你事半功倍的建议。8.1 提示词Prompt管理版本化与A/B测试将提示词存储在数据库或配置文件中而非硬编码在代码里。这允许你动态调整、版本控制和进行A/B测试。模块化设计将系统提示词、工具描述、示例对话few-shot等拆分为不同模块便于维护和复用。持续优化建立反馈循环收集失败案例分析是提示词问题、工具问题还是模型问题并针对性优化。8.2 工具Tool设计单一职责每个工具应只做一件事并做好。避免创建功能臃肿的“瑞士军刀”式工具。防御性编程工具函数内部应有充分的输入验证、异常处理和日志记录。返回给LLM的信息应结构清晰、简洁。成本与延迟考虑对于调用外部API或查询数据库的工具要考虑成本、延迟和失败率。必要时加入缓存、批处理和降级策略。8.3 智能体Agent选型不要迷信“最强大脑”对于流程固定、决策简单的任务使用预定义的链Chain或工作流Workflow往往比通用Agent更可靠、更高效。分层设计复杂系统可以采用“规划器执行器”的分层架构。一个顶层Agent负责规划拆解任务多个下层专用Agent或工具负责执行。人的参与Human-in-the-loop在关键决策点如确认支付、发布内容设置人工审核环节是保障安全性和可靠性的有效手段。8.4 可观测性Observability全链路追踪记录每个请求的唯一ID并贯穿LLM调用、工具调用等所有环节。这有助于问题排查和效果分析。关键指标监控监控Token消耗、请求延迟、工具调用成功率、任务完成率、用户满意度等。会话分析与复盘定期抽样分析对话日志发现智能体的常见失败模式和改进点。8.5 安全与合规输入输出过滤对用户输入和模型输出进行必要的敏感词过滤和内容安全审核。权限控制工具应遵循最小权限原则。例如一个“查询数据”的工具不应拥有“删除数据”的权限。数据隐私明确告知用户数据如何使用避免在提示词中泄露用户隐私信息。考虑对输出进行匿名化处理。9. 总结与展望智能体开发的未来在工程化通过本文的探讨我们可以看到构建一个有用的智能体其挑战已经从“如何让模型理解指令”转变为“如何工程化地构建一个稳定、可靠、可扩展的智能系统”。提示词工程、工具抽象、工作流编排、记忆管理、测试评估这五个支柱构成了现代LLM工程的核心。未来的趋势已经清晰从单智能体到多智能体协作复杂的任务将由多个各司其职的智能体通过通信和协作来完成。从文本交互到多模态感知与行动智能体将能看、能听、能操作图形界面GUI真正成为数字世界的“数字员工”。从规则编排到自主进化结合强化学习等技术智能体可能从人类反馈中学习自主优化其决策策略和工具使用方式。开发工具链的标准化与普及像LangChain、LangGraph、MCP这样的框架和协议正在成为事实标准大幅降低开发门槛。对于开发者而言当下的重点不是追逐最前沿的模型而是扎实掌握这些工程化框架和设计模式。建议你从本文的示例代码出发选择一个具体的业务场景如智能客服、数据分析助手、自动化流程机器人亲手搭建一个智能体原型。在过程中你会更深刻地体会到理论知识与工程实践的结合点。智能体开发不再是少数AI研究者的游戏它正在成为每一位软件工程师工具箱里的新利器。理解其原理掌握其工程方法你就能在AI驱动的未来中构建出真正创造价值的应用。 大语言模型工程实践：从智能体开发到复杂工作流编排
如果你正在开发基于大语言模型LLM的应用是否遇到过这样的困境模型回答看似流畅但一到具体业务场景就“胡说八道”智能体Agent设计时逻辑链一长就失控要么陷入死循环要么忘记核心任务或者团队协作时每个人写的提示词Prompt风格迥异导致效果天差地别难以维护这些问题本质上都不是模型能力的问题而是工程化的问题。大语言模型本身只是一个强大的“大脑”而如何为这个大脑构建稳定、可靠、可协作的“神经系统”和“行动躯体”才是决定AI应用成败的关键。这正是“大语言模型工程”LLM Engineering要解决的核心命题。本文是“大语言模型工程”系列的第二部分。在第一部分我们探讨了基础概念与RAG检索增强生成之后本篇将深入更复杂的领域智能体Agent的开发、工程化实践与前沿架构。我们将不再停留在“调用API”的层面而是聚焦于如何设计一个能自主规划、使用工具、并可靠执行复杂任务的智能系统。你会看到从简单的函数调用到基于LangGraph的工作流编排从提示工程到智能体的“人设”与记忆管理每一步都充满了工程上的权衡与设计。读完本文你将能清晰地构建一个可落地的智能体开发框架理解当前主流工具链如LangChain, LangGraph, MCP的最佳实践并避开智能体开发中最常见的那些“坑”。1. 智能体开发从“玩具”到“工具”的关键跨越很多人对智能体Agent的第一印象可能来自AutoGPT或ChatGPT的插件功能——能自动上网搜索、写代码、分析数据。然而将这些演示转化为真正解决业务问题、稳定运行的生产级应用中间隔着一道巨大的工程鸿沟。一个“玩具级”智能体和“工具级”智能体的核心区别在于可靠性、可控性和可观测性。可靠性能否在百次、千次调用中稳定完成目标而非偶尔成功可控性能否在任务偏离轨道时被干预或纠正能否理解其决策过程可观测性开发者和用户能否清晰地知道智能体“在想什么”、“做了什么”、“为什么失败”例如一个简单的“总结网页内容”任务玩具级智能体可能直接调用搜索工具并生成总结。但工具级智能体需要考虑如果搜索不到结果怎么办如果网页内容过长超出上下文窗口怎么办如果总结出现事实性幻觉Hallucination如何校验这些边界情况的处理正是LLM工程的核心。因此智能体开发远不止是写一段聪明的提示词。它是一个系统工程涉及规划Planning、工具使用Tool Use、记忆Memory、执行Execution和评估Evaluation等多个模块的协同。接下来我们将从基础概念开始逐步拆解如何搭建这样一个系统。2. 核心概念辨析Agent、Tool、Memory与Workflow在深入实践前必须厘清几个容易混淆的核心概念。它们构成了智能体系统的骨架。2.1 智能体Agent vs. 链Chain链Chain是预定义的、线性的执行流程。例如“检索文档 - 生成摘要”是一个链。它的路径是固定的适合确定性高的任务。智能体Agent则具备动态决策能力。它根据当前状态和目标自主决定下一步调用哪个工具甚至是否需要进行多步规划。它适合开放域、路径不确定的复杂任务。简单说链是“铁轨”智能体是“驾驶员”。在LangChain等框架中Agent本质上是一个特殊的Chain它内部封装了决策逻辑通常由一个大语言模型驱动。2.2 工具Tool工具是智能体与外部世界交互的“手”和“脚”。一个工具通常对应一个具体的函数例如search_web(query): 执行网络搜索。execute_sql(sql_query): 查询数据库。call_api(endpoint, params): 调用内部API。read_file(file_path): 读取本地文件。智能体的强大之处在于它能动态组合使用这些工具来完成复杂目标。工程上的关键点在于如何清晰、安全地向LLM描述工具的功能和输入格式。2.3 记忆Memory记忆决定了智能体的“上下文”有多长以及它如何利用历史信息。主要分为两类短期记忆Short-term Memory即对话上下文窗口。通常通过维护一个对话历史列表来实现。长期记忆Long-term Memory超越单次对话周期的信息存储。例如将重要的对话摘要或用户偏好存入向量数据库供后续会话检索。这对于构建有“人设”或持续学习的智能体至关重要。2.4 工作流Workflow与编排Orchestration当单个智能体无法处理过于复杂的任务时我们需要引入工作流。工作流将一个大任务分解为多个子任务并由不同的智能体或节点Node协作完成。编排则是管理这些节点执行顺序和状态流转的机制。近年来LangGraph基于LangChain和MCPModel Context Protocol等框架的兴起正是为了更优雅地解决复杂工作流的编排问题。它们允许你以图Graph的形式定义状态机清晰描述“如果条件A成立则执行B否则执行C”这样的逻辑。3. 环境准备构建你的智能体开发沙箱在开始编码前我们需要搭建一个稳定且高效的开发环境。以下配置是一个兼顾灵活性和生产实践的起点。核心工具栈选择Python 3.10LLM生态的主流语言。LangChain LangGraph当前智能体开发的事实标准框架社区活跃工具链丰富。OpenAI API 或 本地大模型推理核心。为快速原型开发建议使用GPT-4系列API为追求成本可控和数据隐私可部署本地模型如Qwen、DeepSeek或Llama 3。Jupyter Notebook / VS Code交互式开发和调试。向量数据库可选如需长期记忆或RAG能力可准备Chroma、Weaviate或PGVector。3.1 基础环境搭建首先创建并激活一个独立的Python虚拟环境这是管理依赖冲突的最佳实践。# 创建项目目录并进入 mkdir llm-agent-project cd llm-agent-project # 创建虚拟环境以venv为例 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate3.2 依赖安装创建requirements.txt文件包含以下核心依赖# 核心框架 langchain0.1.0 langchain-openai0.0.5 langgraph0.0.30 # 可选用于连接本地模型或其它API langchain-community0.0.10 # 工具与工具调用相关示例 langchain-experimental0.0.50 # 包含一些实验性但有用的Agent # 实用工具 python-dotenv1.0.0 # 管理环境变量 jupyter1.0.0 # 用于交互式探索 # 可选向量数据库客户端以Chroma为例 chromadb0.4.22使用pip进行安装pip install -r requirements.txt3.3 配置API密钥强烈建议使用环境变量管理敏感信息如API Key。创建.env文件并确保将其加入.gitignore# .env 文件示例 OPENAI_API_KEYsk-your-openai-api-key-here # 如果使用其他模型例如通义千问 DASHSCOPE_API_KEYyour-dashscope-key # 如果使用向量数据库 CHROMA_PERSIST_DIRECTORY./chroma_db在代码中通过dotenv加载# config.py import os from dotenv import load_dotenv load_dotenv() OPENAI_API_KEY os.getenv(OPENAI_API_KEY) if not OPENAI_API_KEY: raise ValueError(请在 .env 文件中设置 OPENAI_API_KEY)至此你的智能体开发沙箱就准备就绪了。4. 从零构建你的第一个智能体一个任务规划助手让我们从一个具体的例子开始构建一个“周末活动规划助手”。这个智能体需要根据用户的简单需求如“我想在周末做些放松又有趣的事预算中等”自动调用工具来搜索信息、评估选项并生成一个可行的计划。4.1 第一步定义工具Tools工具是智能体的能力边界。我们定义三个简单的工具# tools.py from langchain.tools import tool from typing import Optional tool def search_local_events(city: str, date: str, category: Optional[str] None) - str: 根据城市、日期和可选类别搜索本地活动如展览、音乐会、市集。 这是一个模拟函数实际项目中应接入真实数据API。 # 模拟数据 events_db { 北京: { 2024-05-25: [故宫博物院特展, 798艺术区新潮市集, 国家大剧院音乐会], 2024-05-26: [奥林匹克公园骑行活动, 三里屯咖啡文化节] }, 上海: { 2024-05-25: [上海当代艺术博物馆展览, 迪士尼乐园特别活动], 2024-05-26: [外滩源周末市集, 浦东图书馆讲座] } } events events_db.get(city, {}).get(date, []) if category: # 简单模拟类别过滤 events [e for e in events if category in e] return f在{city}{date}找到的活动{, .join(events) if events else 暂无活动} tool def get_weather(city: str, date: str) - str: 获取指定城市和日期的天气预报。 # 模拟天气数据 weather_data { 北京: {2024-05-25: 晴15-25°C, 2024-05-26: 多云18-27°C}, 上海: {2024-05-25: 小雨20-28°C, 2024-05-26: 阴22-30°C} } forecast weather_data.get(city, {}).get(date, 暂无预报信息) return f{city}{date}的天气{forecast} tool def estimate_budget(activity: str, people: int 2) - str: 估算某项活动的大致人均预算单位元。 # 简单的预算估算逻辑 budget_map { 展览: 100, 音乐会: 300, 市集: 50, 骑行: 20, 咖啡: 80, 讲座: 0, 迪士尼: 500 } base_cost 0 for key, cost in budget_map.items(): if key in activity: base_cost cost break total base_cost * people return f活动「{activity}」{people}人参与预估总花费约{total}元人均{base_cost}元。4.2 第二步创建智能体Agent我们将使用LangChain的create_react_agent这是一种基于“Reasoning Acting”模式的强大智能体。# agent_basic.py from langchain import hub from langchain.agents import create_react_agent, AgentExecutor from langchain_openai import ChatOpenAI from tools import search_local_events, get_weather, estimate_budget from config import OPENAI_API_KEY # 1. 初始化大语言模型 llm ChatOpenAI( modelgpt-4-turbo-preview, # 对于复杂规划任务建议使用能力更强的模型 temperature0.2, # 较低的温度使输出更稳定、更可预测 api_keyOPENAI_API_KEY ) # 2. 准备工具列表 tools [search_local_events, get_weather, estimate_budget] # 3. 从LangChain Hub获取一个优化的ReAct提示词模板 # 提示词工程是Agent性能的关键使用社区验证过的模板是很好的起点。 prompt hub.pull(hwchase17/react) # 4. 创建ReAct Agent agent create_react_agent(llm, tools, prompt) # 5. 创建Agent执行器它负责运行Agent并处理工具调用 agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, # 开启详细日志便于调试和理解Agent的思考过程 handle_parsing_errorsTrue, # 优雅处理模型输出解析错误 max_iterations5, # 限制最大迭代次数防止无限循环 early_stopping_methodgenerate # 当Agent认为任务完成时可以提前停止 )4.3 第三步运行与测试现在让我们用这个智能体来回答一个用户查询。# 运行智能体 if __name__ __main__: query 我这个周末2024-05-25在北京想找一些预算在每人200元以内的户外活动请帮我规划一下并考虑天气因素。 print(f用户提问: {query}\n) print(*50) try: result agent_executor.invoke({input: query}) print(\n *50) print(最终回答:) print(result[output]) except Exception as e: print(f执行过程中出现错误: {e})预期输出与过程解析verboseTrue时用户提问: 我这个周末2024-05-25在北京想找一些预算在每人200元以内的户外活动请帮我规划一下并考虑天气因素。 进入新的Agent执行链... 思考用户在北京日期是2024-05-25需要户外活动人均预算200以内。我需要先查天气再找活动最后估算预算。 行动get_weather 行动输入{city: 北京, date: 2024-05-25} 观察北京2024-05-25的天气晴15-25°C 思考天气很好适合户外。现在搜索北京当天的活动。 行动search_local_events 行动输入{city: 北京, date: 2024-05-25, category: 户外} 观察在北京2024-05-25找到的活动奥林匹克公园骑行活动 思考找到了一个户外活动。现在需要估算其预算是否满足要求。 行动estimate_budget 行动输入{activity: 奥林匹克公园骑行活动, people: 1} 观察活动「奥林匹克公园骑行活动」1人参与预估总花费约20元人均20元。 思考人均20元远低于200元预算符合要求。我可以基于这些信息给出建议了。 最终回答: 根据您的要求我为您规划了2024-05-25周六在北京的活动 1. **活动建议**奥林匹克公园骑行活动。这是一个户外活动非常适合晴朗的天气当天预报晴15-25°C。 2. **预算分析**该活动人均花费约20元远低于您设定的200元预算。 3. **综合建议**这是一个性价比很高的选择。建议您准备好骑行装备并做好防晒。由于是户外活动请根据实时天气情况调整着装。通过verboseTrue的日志你可以清晰地看到智能体的“思考-行动-观察”循环ReAct Loop这是理解和调试Agent行为的最重要窗口。5. 进阶使用LangGraph构建可控的复杂工作流基础的Agent能处理简单任务但对于需要严格步骤、条件分支或多人协作的复杂流程我们就需要更强大的编排能力。LangGraph允许我们将工作流定义为一个有向图Graph其中节点是函数或子智能体边定义了执行路径。让我们构建一个更复杂的“智能旅行规划师”它包含决策分支和状态管理。5.1 定义状态State状态是贯穿整个工作流的数据容器。# travel_planner_graph.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END # 1. 定义状态结构 class TravelPlanState(TypedDict): 旅行规划的工作流状态 user_request: str # 原始用户请求 destination: str # 解析出的目的地 dates: str # 解析出的日期 traveler_count: int # 旅行人数 budget_constraint: str # 预算约束 # 中间结果 weather_info: str flight_options: List[str] hotel_options: List[str] activity_options: List[str] # 最终输出 final_itinerary: str warnings: List[str]5.2 创建节点Nodes每个节点是一个处理特定任务的函数。# 2. 定义各个节点函数 def parse_user_request(state: TravelPlanState) - TravelPlanState: 节点1解析用户请求提取关键信息 from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0) prompt ChatPromptTemplate.from_messages([ (system, 你是一个旅行请求解析助手。从用户请求中提取目的地、日期、人数和预算信息。), (human, {request}) ]) chain prompt | llm # 为简化示例这里直接模拟解析结果 # 实际应用中应使用Pydantic或类似方法进行结构化输出 state[destination] 上海 state[dates] 2024-06-01 至 2024-06-03 state[traveler_count] 2 state[budget_constraint] 中等预算 print(f[节点-解析] 已解析请求: 目的地{state[destination]}, 日期{state[dates]}) return state def check_weather_and_warn(state: TravelPlanState) - TravelPlanState: 节点2检查目的地天气并生成警告 # 模拟天气查询 state[weather_info] 上海2024-06-01至2024-06-03多云转小雨气温22-30°C。 if 小雨 in state[weather_info]: state.setdefault(warnings, []).append(出行期间可能有雨建议携带雨具。) print(f[节点-天气] 天气信息已获取: {state[weather_info]}) return state def search_flights(state: TravelPlanState) - TravelPlanState: 节点3搜索航班模拟 # 模拟航班搜索实际应调用API state[flight_options] [ 航班A: 北京-上海 08:00-10:15 经济舱 ¥1200, 航班B: 北京-上海 14:30-16:45 经济舱 ¥950 ] print(f[节点-航班] 找到 {len(state[flight_options])} 个航班选项) return state def search_hotels(state: TravelPlanState) - TravelPlanState: 节点4搜索酒店模拟 state[hotel_options] [ 酒店X: 市中心 四星级 ¥600/晚 评分4.5, 酒店Y: 近景区 三星级 ¥400/晚 评分4.2 ] print(f[节点-酒店] 找到 {len(state[hotel_options])} 个酒店选项) return state def search_activities(state: TravelPlanState) - TravelPlanState: 节点5搜索当地活动模拟 state[activity_options] [ 外滩观光, 迪士尼乐园一日游, 豫园文化体验 ] print(f[节点-活动] 找到 {len(state[activity_options])} 个活动选项) return state def generate_itinerary(state: TravelPlanState) - TravelPlanState: 节点6整合所有信息生成最终行程计划 itinerary f 上海旅行计划{state[dates]} **出行人数:** {state[traveler_count]}人 **预算水平:** {state[budget_constraint]} **天气情况:** {state[weather_info]} **推荐航班:** {chr(10).join(f- {f} for f in state.get(flight_options, []))} **推荐酒店:** {chr(10).join(f- {h} for h in state.get(hotel_options, []))} **推荐活动:** {chr(10).join(f- {a} for a in state.get(activity_options, []))} **温馨提示:** {chr(10).join(f- {w} for w in state.get(warnings, [])) if state.get(warnings) else 无} state[final_itinerary] itinerary print([节点-生成] 行程计划已生成) return state5.3 构建图Graph并编排流程# 3. 构建工作流图 def build_travel_planner_graph(): # 初始化图 workflow StateGraph(TravelPlanState) # 添加节点 workflow.add_node(parse_request, parse_user_request) workflow.add_node(check_weather, check_weather_and_warn) workflow.add_node(search_flights, search_flights) workflow.add_node(search_hotels, search_hotels) workflow.add_node(search_activities, search_activities) workflow.add_node(generate_itinerary, generate_itinerary) # 定义边执行顺序 workflow.set_entry_point(parse_request) workflow.add_edge(parse_request, check_weather) # 并行执行航班、酒店、活动搜索 workflow.add_edge(check_weather, search_flights) workflow.add_edge(check_weather, search_hotels) workflow.add_edge(check_weather, search_activities) # 等待所有并行节点完成后进入生成节点 workflow.add_edge(search_flights, generate_itinerary) workflow.add_edge(search_hotels, generate_itinerary) workflow.add_edge(search_activities, generate_itinerary) workflow.add_edge(generate_itinerary, END) # 编译图 return workflow.compile() # 4. 运行工作流 if __name__ __main__: # 初始化状态 initial_state: TravelPlanState { user_request: 我想六月初去上海玩三天两个人中等预算。, destination: , dates: , traveler_count: 0, budget_constraint: , weather_info: , flight_options: [], hotel_options: [], activity_options: [], final_itinerary: , warnings: [] } print(开始执行旅行规划工作流...\n) app build_travel_planner_graph() final_state app.invoke(initial_state) print(\n *60) print(工作流执行完成最终行程计划) print(*60) print(final_state[final_itinerary])这个例子展示了LangGraph的核心优势显式的工作流定义。你可以清晰地看到每个步骤节点和它们的依赖关系。对于需要严格SOP标准作业程序的业务流程如图中所示的并行搜索这种基于图的编排比传统线性Agent可靠得多。6. 工程化实践提升智能体的可靠性与可维护性将智能体从Demo推向生产必须解决可靠性、可维护性和团队协作问题。以下是几个关键的工程实践。6.1 结构化输出Structured Outputs让LLM返回结构化数据如JSON、Pydantic模型而非自由文本是后续程序化处理的基础。这能极大减少解析错误。# structured_agent.py from langchain_openai import ChatOpenAI from langchain_core.pydantic_v1 import BaseModel, Field from langchain_core.prompts import ChatPromptTemplate from typing import List # 1. 定义期望的输出结构 class Itinerary(BaseModel): 行程计划的数据模型 destination: str Field(description旅行目的地) start_date: str Field(description开始日期) end_date: str Field(description结束日期) recommended_flights: List[str] Field(description推荐的航班列表) total_estimated_cost: float Field(description预估总花费) notes: List[str] Field(description注意事项) # 2. 创建支持结构化输出的链 llm ChatOpenAI(modelgpt-4-turbo-preview) structured_llm llm.with_structured_output(Itinerary) # 关键方法 prompt ChatPromptTemplate.from_messages([ (system, 你是一个旅行规划专家。根据用户请求生成结构化的行程计划。), (human, 用户请求{request}) ]) structured_chain prompt | structured_llm # 3. 调用并获取结构化对象 result structured_chain.invoke({ request: 我想下个月去杭州玩两天预算5000元左右。 }) print(f目的地: {result.destination}) print(f日期: {result.start_date} 至 {result.end_date}) print(f预估花费: ¥{result.total_estimated_cost}) print(注意事项:) for note in result.notes: print(f - {note})6.2 智能体的人设Persona与系统提示词工程系统提示词System Prompt是智能体的“人设”和“行为准则”。一个好的提示词能显著提升输出质量和稳定性。# persona_prompt.py from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from tools import search_local_events, get_weather # 复用之前的工具 # 定义一个专业、细致、谨慎的“旅行顾问”人设 system_prompt 你是一位资深的旅行规划顾问名叫“智行”。你的性格特点是 1. **专业严谨**提供的所有信息如价格、时间、地址都必须力求准确不确定时要明确说明。 2. **考虑周全**主动考虑用户未提及的因素如天气、交通、季节性活动、安全提醒。 3. **预算敏感**始终将用户的预算约束放在首位提供不同价位的选择并对比优劣。 4. **主动确认**在做出关键建议如预订类项目前应询问用户是否需要更详细的信息或确认。 5. **诚实透明**如果某些信息无法获取或工具调用失败如实告知用户而不是编造。 请根据以上原则调用合适的工具来回答用户的问题。 prompt ChatPromptTemplate.from_messages([ (system, system_prompt), MessagesPlaceholder(variable_namechat_history, optionalTrue), # 支持对话历史 (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), # Agent思考过程占位符 ]) llm ChatOpenAI(modelgpt-4-turbo, temperature0.3) tools [search_local_events, get_weather] # 使用OpenAI最新的tools calling格式创建Agent agent create_openai_tools_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) # 测试 result agent_executor.invoke({ input: 我周末想去户外走走但听说可能会下雨有什么建议吗, chat_history: [] # 如果是连续对话这里传入历史消息 })6.3 记忆Memory管理为了让智能体在长对话中保持连贯需要有效管理记忆。LangChain提供了多种记忆方案。# memory_management.py from langchain.memory import ConversationBufferWindowMemory, CombinedMemory, ConversationSummaryMemory from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI # 组合多种记忆策略 memory CombinedMemory( memories[ ConversationBufferWindowMemory( memory_keychat_history, # 存储最近几轮对话原文 k5, # 保留最近5轮对话 input_keyinput, output_keyoutput ), ConversationSummaryMemory( memory_keysummary_memory, # 存储对话摘要用于长期上下文 llmChatOpenAI(modelgpt-3.5-turbo, temperature0), input_keyinput ) ] ) # 在创建Agent时传入memory llm ChatOpenAI(modelgpt-4-turbo) tools [...] # 你的工具列表 prompt ... # 包含chat_history和summary_memory占位符的提示词 agent create_openai_tools_agent(llm, tools, prompt) agent_executor AgentExecutor( agentagent, toolstools, memorymemory, verboseTrue ) # 进行多轮对话 agent_executor.invoke({input: 我喜欢看科幻电影。}) agent_executor.invoke({input: 根据我的喜好推荐一部最近上映的。}) # Agent能记住上一轮的喜好6.4 测试与评估Evaluation如何衡量一个智能体的好坏不能只靠人工看几次对话。需要建立自动化的评估体系。# agent_evaluation.py import asyncio from langchain.evaluation import load_evaluator from langchain.evaluation import EvaluatorType # 1. 准备测试用例 test_cases [ { input: 北京明天天气怎么样, expected_actions: [get_weather] # 期望调用的工具 }, { input: 帮我规划一个预算500元的北京一日游, expected_actions: [get_weather, search_local_events, estimate_budget], expected_keywords: [预算, 500, 推荐] # 期望回答中包含的关键词 } ] # 2. 使用LangChain内置评估器 evaluator load_evaluator(EvaluatorType.QA) # 也可以使用CriteriaEvalChain来评估是否符合特定标准如“友好性”、“相关性” async def evaluate_agent(agent_executor, test_case): 异步评估单个测试用例 try: result await agent_executor.ainvoke({input: test_case[input]}) actual_output result[output] # 简单评估检查是否调用了期望的工具可通过日志解析 # 以及输出是否包含关键词 score 0 for keyword in test_case.get(expected_keywords, []): if keyword in actual_output: score 1 # 更复杂的评估可以使用LLM作为裁判 # eval_result await evaluator.aevaluate_strings( # predictionactual_output, # inputtest_case[input], # referencetest_case.get(expected_output, ) # ) return { input: test_case[input], output: actual_output, score: score, passed: score len(test_case.get(expected_keywords, [])) * 0.8 # 80%关键词命中即通过 } except Exception as e: return {input: test_case[input], error: str(e), passed: False} # 3. 批量运行评估 async def run_evaluations(): tasks [evaluate_agent(agent_executor, tc) for tc in test_cases] results await asyncio.gather(*tasks) passed sum(1 for r in results if r.get(passed, False)) print(f评估完成。通过率: {passed}/{len(results)} ({passed/len(results)*100:.1f}%)) for r in results: print(f输入: {r[input][:50]}... | 通过: {r.get(passed, N/A)} | 得分: {r.get(score, N/A)}) # 注意需要在一个异步环境中运行 # asyncio.run(run_evaluations())7. 常见问题与排查指南在智能体开发过程中你几乎一定会遇到以下问题。这里提供快速的排查思路。问题现象可能原因排查方式解决方案Agent陷入循环反复调用同一工具或思考。1. 提示词未明确停止条件。2. 工具返回信息不足以让LLM做出决策。3.max_iterations设置过高。1. 查看verboseTrue日志观察思考过程。2. 检查工具返回内容是否清晰、格式是否标准。1. 在系统提示词中明确“当你拥有足够信息时请直接给出最终答案”。2. 优化工具返回格式增加明确的状态标识如“搜索完成共X条结果”。3. 合理设置max_iterations通常5-10次。工具调用参数错误如类型不匹配或缺少参数。1. 工具函数描述docstring不清晰。2. LLM未能正确理解工具输入格式。1. 检查工具函数的类型注解和docstring是否准确描述了每个参数。2. 使用agent_executor.invoke并捕获异常查看具体错误。1. 使用Pydantic模型严格定义工具输入。2. 在提示词中举例说明工具调用格式。3. 使用OpenAI的tools格式而非旧的bind_tools其参数解析更准确。LLM不调用工具直接基于自身知识回答。1. 问题太简单LLM认为自己能直接回答。2. 工具描述不够有吸引力或必要。1. 询问一个必须调用工具才能回答的问题如“今天上海天气如何”。2. 检查工具描述是否强调了其独特价值如“获取实时数据”。1. 在系统提示词中强调“你必须使用工具来获取最新或特定信息”。2. 对简单问题可以接受LLM直接回答对需要准确性的问题强制其调用工具。响应速度慢。1. 工具本身是慢速IO操作如网络请求。2. LLM生成速度慢大模型或网络延迟。3. Agent迭代次数过多。1. 使用日志记录每个工具调用耗时。2. 检查网络连接和API响应时间。1. 为慢速工具设置超时timeout和重试机制。2. 考虑使用更快的LLM如GPT-3.5-Turbo进行简单决策。3. 使用异步调用ainvoke并行执行独立工具。生产环境部署后不稳定。1. API密钥、网络等环境问题。2. 提示词或工具版本变更。3. 流量突增导致限流。1. 实现完善的日志和监控记录每次调用的输入、输出、耗时和错误。2. 进行版本化管理每次变更前在测试环境充分验证。1. 增加健壮的错误处理retry, fallback。2. 使用配置中心管理提示词和参数。3. 实现限流和熔断机制保护下游服务。8. 最佳实践与架构建议基于大量的项目经验以下是一些能让你事半功倍的建议。8.1 提示词Prompt管理版本化与A/B测试将提示词存储在数据库或配置文件中而非硬编码在代码里。这允许你动态调整、版本控制和进行A/B测试。模块化设计将系统提示词、工具描述、示例对话few-shot等拆分为不同模块便于维护和复用。持续优化建立反馈循环收集失败案例分析是提示词问题、工具问题还是模型问题并针对性优化。8.2 工具Tool设计单一职责每个工具应只做一件事并做好。避免创建功能臃肿的“瑞士军刀”式工具。防御性编程工具函数内部应有充分的输入验证、异常处理和日志记录。返回给LLM的信息应结构清晰、简洁。成本与延迟考虑对于调用外部API或查询数据库的工具要考虑成本、延迟和失败率。必要时加入缓存、批处理和降级策略。8.3 智能体Agent选型不要迷信“最强大脑”对于流程固定、决策简单的任务使用预定义的链Chain或工作流Workflow往往比通用Agent更可靠、更高效。分层设计复杂系统可以采用“规划器执行器”的分层架构。一个顶层Agent负责规划拆解任务多个下层专用Agent或工具负责执行。人的参与Human-in-the-loop在关键决策点如确认支付、发布内容设置人工审核环节是保障安全性和可靠性的有效手段。8.4 可观测性Observability全链路追踪记录每个请求的唯一ID并贯穿LLM调用、工具调用等所有环节。这有助于问题排查和效果分析。关键指标监控监控Token消耗、请求延迟、工具调用成功率、任务完成率、用户满意度等。会话分析与复盘定期抽样分析对话日志发现智能体的常见失败模式和改进点。8.5 安全与合规输入输出过滤对用户输入和模型输出进行必要的敏感词过滤和内容安全审核。权限控制工具应遵循最小权限原则。例如一个“查询数据”的工具不应拥有“删除数据”的权限。数据隐私明确告知用户数据如何使用避免在提示词中泄露用户隐私信息。考虑对输出进行匿名化处理。9. 总结与展望智能体开发的未来在工程化通过本文的探讨我们可以看到构建一个有用的智能体其挑战已经从“如何让模型理解指令”转变为“如何工程化地构建一个稳定、可靠、可扩展的智能系统”。提示词工程、工具抽象、工作流编排、记忆管理、测试评估这五个支柱构成了现代LLM工程的核心。未来的趋势已经清晰从单智能体到多智能体协作复杂的任务将由多个各司其职的智能体通过通信和协作来完成。从文本交互到多模态感知与行动智能体将能看、能听、能操作图形界面GUI真正成为数字世界的“数字员工”。从规则编排到自主进化结合强化学习等技术智能体可能从人类反馈中学习自主优化其决策策略和工具使用方式。开发工具链的标准化与普及像LangChain、LangGraph、MCP这样的框架和协议正在成为事实标准大幅降低开发门槛。对于开发者而言当下的重点不是追逐最前沿的模型而是扎实掌握这些工程化框架和设计模式。建议你从本文的示例代码出发选择一个具体的业务场景如智能客服、数据分析助手、自动化流程机器人亲手搭建一个智能体原型。在过程中你会更深刻地体会到理论知识与工程实践的结合点。智能体开发不再是少数AI研究者的游戏它正在成为每一位软件工程师工具箱里的新利器。理解其原理掌握其工程方法你就能在AI驱动的未来中构建出真正创造价值的应用。