LangGraph实战用Doubao模型和Tavily搜索构建会思考的AI助手想象一下当你向AI助手提出2024年巴黎奥运会100米自由泳决赛冠军的家乡是哪里这样的问题时一个普通的聊天机器人可能会直接搜索并返回一个可能不准确的答案。而一个真正智能的助手会先思考要回答这个问题我需要先找到冠军是谁然后再查询这个人的家乡信息——这就是规划执行型智能体的核心能力。1. 为什么需要会先想后做的AI助手在现实世界的复杂问题解决中线性思维往往不够用。传统AI助手在处理多步骤查询时容易出现以下问题跳跃式回答直接猜测最终答案而缺少中间验证步骤信息碎片化无法将多个信息源的结果有机整合缺乏适应性遇到意外情况时不会调整策略Doubao模型的doubao-1-5-thinking-pro-250415版本特别擅长任务分解与规划配合LangGraph的状态管理可以构建出真正会思考的工作流。以下是传统AI与规划执行型智能体的对比对比维度传统AI助手规划执行型智能体问题处理方式直接回答先规划后执行多步骤查询容易出错分步验证异常处理僵化动态调整可解释性低高可见思考过程2. 环境配置与工具链整合2.1 准备工作环境首先确保你的开发环境已准备就绪# 创建并激活虚拟环境 python -m venv langgraph-env source langgraph-env/bin/activate # Linux/Mac # langgraph-env\Scripts\activate # Windows # 安装核心依赖 pip install langgraph langchain tavily-python python-dotenv提示建议使用Python 3.10版本以获得最佳异步支持2.2 API密钥管理安全地管理你的API凭证是第一步。创建一个.env文件存储敏感信息# .env文件示例 DOUBAO_API_KEYyour_actual_key_here TAVILY_API_KEYyour_actual_key_here然后在代码中通过环境变量加载from dotenv import load_dotenv import os load_dotenv() doubao_api_key os.getenv(DOUBAO_API_KEY) tavily_api_key os.getenv(TAVILY_API_KEY)3. 构建智能体的核心组件3.1 初始化Doubao模型doubao-1-5-thinking-pro-250415版本在任务规划方面表现出色from langchain_openai import ChatOpenAI llm ChatOpenAI( base_urlhttps://ark.cn-beijing.volces.com/api/v3, api_keydoubao_api_key, modeldoubao-1-5-thinking-pro-250415, temperature0.3 # 适度创造性 )3.2 配置Tavily搜索工具Tavily提供高质量的API搜索服务特别适合AI应用from langchain_community.tools.tavily_search import TavilySearchResults search TavilySearchResults( max_results3, # 平衡质量与效率 tavily_api_keytavily_api_key )注意设置合理的max_results可以避免信息过载通常3-5个优质结果比大量低质结果更有价值4. 设计智能体工作流4.1 定义状态数据结构使用TypedDict清晰地定义智能体的思考状态from typing import TypedDict, List, Tuple import operator class AgentState(TypedDict): input: str # 用户原始问题 plan: List[str] # 执行步骤列表 past_steps: Annotated[List[Tuple], operator.add] # 已完成步骤及结果 response: str # 最终响应4.2 创建规划器组件规划器负责将复杂问题分解为可执行步骤from langchain_core.prompts import ChatPromptTemplate from langchain_core.pydantic_v1 import BaseModel, Field class Plan(BaseModel): steps: List[str] Field(description按顺序排列的执行步骤) planner_prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的问题分解专家。对于给定的任务 1. 分析需要哪些信息才能准确回答 2. 拆解为最小必要步骤 3. 确保最后一步能得到最终答案 4. 不要假设任何未知信息), (user, {input}) ]) planner planner_prompt | llm.with_structured_output(Plan)5. 实现执行与动态调整5.1 构建执行节点执行节点负责处理具体的子任务async def execute_step(state: AgentState): current_step state[plan][0] step_instruction f 当前执行计划 {chr(10).join(f{i1}. {step} for i, step in enumerate(state[plan]))} 现在需要执行第1步{current_step} result await search.ainvoke({query: step_instruction}) return {past_steps: [(current_step, result)]}5.2 动态重新规划机制智能体需要根据新信息调整策略from typing import Union class ReplanDecision(BaseModel): action: Union[str, List[str]] Field( description继续执行原计划/返回最终答案/调整后续步骤 ) replanner_prompt ChatPromptTemplate.from_template( 当前目标{input} 原始计划{plan} 已完成步骤及结果{past_steps} 请决定下一步行动 1. 如果已获得足够信息直接给出最终答案 2. 如果计划仍适用返回continue 3. 如果需要调整计划返回新的步骤列表 ) replanner replanner_prompt | llm.with_structured_output(ReplanDecision)6. 组装完整工作流6.1 构建状态图使用LangGraph将各组件连接成有机整体from langgraph.graph import StateGraph, END workflow StateGraph(AgentState) # 添加节点 workflow.add_node(plan, planner.ainvoke) workflow.add_node(execute, execute_step) workflow.add_node(replan, replanner.ainvoke) # 设置边 workflow.add_edge(plan, execute) workflow.add_edge(execute, replan) # 条件边 def decide_next_step(state): decision state.get(replan_result) if isinstance(decision, str): return END elif decision continue: return execute else: state[plan] decision return execute workflow.add_conditional_edges(replan, decide_next_step) workflow.set_entry_point(plan) agent workflow.compile()6.2 流式执行示例体验智能体的思考过程async def run_agent(question): async for event in agent.astream({input: question}): if plan in event: print(f 制定计划{event[plan]}) elif execute in event: print(f 执行步骤{event[execute][0]}) print(f 获得结果{event[execute][1][:100]}...) elif replan in event: print( 评估进展...) # 运行示例 query 特斯拉最新款电动车的续航里程比前代提升了多少百分比 await run_agent(query)7. 高级技巧与优化建议7.1 提升规划质量的方法分步验证为规划器添加验证环节validator_prompt 请检查以下计划是否合理 1. 每个步骤是否明确可执行 2. 步骤顺序是否符合逻辑 3. 是否遗漏了必要环节 计划{plan} 多方案投票生成3个候选计划让模型选择最优7.2 性能优化策略并行执行对无依赖关系的步骤启用并行from langgraph.graph import CONCURRENT workflow.add_edge(split, step1) workflow.add_edge(split, step2) workflow.add_edge(step1, merge) workflow.add_edge(step2, merge) workflow.add_edge(merge, END)缓存中间结果避免重复执行相同查询超时控制为每个步骤设置合理时限from asyncio import TimeoutError try: result await asyncio.wait_for(step_execution, timeout10.0) except TimeoutError: # 处理超时情况7.3 调试与监控实现一个简单的日志中间件class ExecutionLogger: async def __call__(self, input): print(f→ 输入{input[input]}) result await self.next.ainvoke(input) print(f← 输出{result}) return result # 添加到执行链路 logged_agent ExecutionLogger() | agent8. 真实案例技术调研助手让我们构建一个能自动完成技术调研的智能体tech_agent workflow.compile() async def research_technology(topic): question f请系统调研{topic}技术 1. 核心原理与架构 2. 主要优缺点 3. 典型应用场景 4. 学习资源推荐 result await tech_agent.ainvoke({input: question}) return result[response] # 示例调研RAG技术 report await research_technology(检索增强生成(RAG)) print(report)这个智能体会自动规划调研框架分步骤查询各项内容整合最终报告遇到信息冲突时会进行验证查询9. 避坑指南在开发过程中我们遇到过几个典型问题过度规划模型有时会生成过多不必要的步骤解决方案添加步骤数量限制如最多5步信息丢失在多步骤中关键数据被遗忘解决方案设计状态时包含关键信息缓存API不稳定第三方服务可能超时解决方案实现自动重试机制from tenacity import retry, stop_after_attempt retry(stopstop_after_attempt(3)) async def reliable_search(query): return await search.ainvoke(query)模型固执有时坚持错误假设解决方案在提示中强调验证意识10. 扩展应用场景这种架构可适配多种专业领域金融分析自动收集数据→生成报告医疗咨询症状分析→鉴别诊断→建议电商客服问题分类→分步解决→推荐关键调整点领域特定的工具集如财经API专业术语增强的提示词行业规范的结果格式化class MedicalReport(BaseModel): symptoms: List[str] possible_conditions: List[str] recommended_actions: List[str] medical_planner planner_prompt.partial( system_message你是一位资深医疗顾问需要... ) | llm.with_structured_output(MedicalReport)11. 架构演进方向随着需求复杂化可以考虑分层规划将大任务分解为子任务树多专家协同不同步骤由专用模型处理长期记忆保存历史决策供未来参考人工监督关键节点请求确认一个进阶架构示例graph TD A[用户输入] -- B{复杂度判断} B --|简单| C[直接回答] B --|复杂| D[主规划器] D -- E[子任务1] D -- F[子任务2] E -- G[子专家A] F -- H[子专家B] G -- I[结果整合] H -- I I -- J[最终审核] J -- K[用户输出]注意实际开发中应避免过度设计从最小可行产品开始迭代12. 效果评估指标要科学评估智能体性能可以跟踪指标测量方式目标值任务完成率成功解决的问题占比85%步骤效率平均每个问题所需步骤2-5步准确率结果人工验证正确率90%响应时间端到端延迟15秒异常率需要人工干预的比例5%建立自动化测试集test_cases [ { input: 苹果Vision Pro的主要技术参数是什么, expected: [micro-OLED, 分辨率, 视场角], min_steps: 2 }, # 更多测试案例... ] async def run_test_suite(): results [] for case in test_cases: start time.time() response await agent.ainvoke(case[input]) elapsed time.time() - start passed all( keyword in response[response] for keyword in case[expected] ) results.append({ input: case[input], passed: passed, time: elapsed }) return results13. 持续改进策略基于实际使用数据的迭代方法收集失败案例建立错误样本库根本原因分析规划缺陷40%执行错误30%信息整合问题20%其他10%针对性增强为常见错误模式添加防护规则优化关键步骤的提示词增加验证环节一个实用的调试技巧——思维回放async def debug_agent(question): history [] async def traced_execution(input): history.append({step: input, data: input}) result await agent.ainvoke(input) history.append({step: output, data: result}) return result await traced_execution({input: question}) # 生成可读报告 report [] for item in history: if item[step] input: report.append(f用户问题{item[data][input]}) else: if plan in item[data]: report.append(制定的计划) report.extend(item[data][plan]) return \n.join(report)14. 安全与合规考量在构建生产级应用时需注意内容过滤对最终输出进行安全检查API限流避免突发流量导致服务中断隐私保护不记录敏感用户数据成本控制监控API调用消耗实现一个简单的成本监控class CostMonitor: def __init__(self): self.counter 0 async def track(self, call): self.counter 1 # 假设每次LLM调用成本为0.01单位 cost 0.01 * self.counter if cost 1.0: # 设置预算警报 warn(API成本已超过1.0单位) return await call # 包装原有执行链 monitored_agent CostMonitor().track(agent.ainvoke)15. 前沿探索方向当前最值得关注的几个演进方向动态工具注册运行时发现和集成新API多模态扩展处理图像、音频等非文本信息自我优化从历史交互中学习改进策略情感智能识别和适应用户情绪状态一个实验性的自我反思机制实现class SelfReflectingAgent: async def __call__(self, input): result await agent.ainvoke(input) analysis await llm.ainvoke( f评估刚才的回答质量 问题{input} 回答{result} 请指出可以改进的三个方面 ) self.improve_based_on(analysis) return result在实际项目中我们发现最实用的优化往往来自对真实用户交互的分析。例如当用户问如何选择适合的云服务时优秀的智能体会先询问预算和使用场景而不是直接列举选项。这种交互式澄清的能力是区分普通工具和真正智能助手的关键。