☰
Agent的自主规划和工具开发
2026/11/6 6:41:24 网站建设 项目流程

目录

一、什么是Agent

二、Agent框架介绍

三、Agent分类

1、反应式(Reactive)

CASE:私募基金运作指引问答助手

2、深思熟虑(Deliberative)

CASE:智能投研助手

3、混合式(Hybrid)

CASE:投顾AI助手


在本篇的开头,我们复习一下Agent的相关内容 初识Agent

一、什么是Agent

  • 大模型是大脑,Agent 是工程系统
    DeepSeek/Kimi 这类大模型本身更像“大脑”,主要负责推理:给输入,产出判断或回答。但单个大脑不等于 Agent;Agent 是把大模型放进一个工程框架里,给它配上规划、工具、记忆等组件。

  • Agent 的基本结构:Planning → Action → Memory
    复杂任务不能只想一步做一步,而要先规划再执行。Planning 负责拆解和组织步骤;Action 对应调用工具完成任务;Memory 为规划和行动提供上下文。

  • Memory 分短期和长期

    • 短期记忆:还留在当前上下文/内存里的信息,典型就是多轮对话内容。

    • 长期记忆:不依赖当前上下文,写到文件、数据库或“小本本”里,之后需要时再主动查找、读取。

记忆方式核心原理存储内容形态优点缺点适用场景
BufferMemory(全量缓存)完整保存所有对话轮次,不做任何压缩原始对话全文信息完整、精确无误,实现最简单Token 随对话增长无限消耗,最终超出模型上下文窗口短对话、Demo 测试、轮次很少的聊天
BufferWindowMemory(滑动窗口)只保留最近 K 轮对话,更早内容直接丢弃,k参数控制保留轮数最近 K 轮的对话全文Token 开销稳定可控超出 K 轮的历史直接丢失,无法回忆早期内容话题切换频繁的一般聊天、普通客服对话
ConversationMemory(会话/摘要记忆)由 LLM 定期把长对话压缩为摘要,摘要随每轮对话更新压缩后的摘要文本相比全量缓存大幅减少 Token 占用需要额外 LLM 调用成本,且压缩过程可能丢失细节主题连贯的长对话
VectorStorebackedMemory(向量库长期记忆)将对话片段向量化后存入向量数据库(FAISS、Chroma 等),检索时按语义相似度而非时间顺序召回向量化的对话片段模拟人类"联想记忆",可从海量历史中精准抓取相关信息,支持跨会话长期记忆需额外维护向量库与嵌入调用;召回结果可能不连续,上下文时间线易断裂(此点检索结果未直接说明,属一般性推断)个人专属助手、需要跨会话长期记忆的系统
  • Agent 的实现通常落在 SDK/工具箱里
    这些组件会被封装进工程框架,例如 LangGraph、千问 Agent、LlamaIndex 等,用来快速搭建具备规划、工具调用和记忆能力的 Agent。

二、Agent框架介绍

Agent的开发框架分为以下三种:

1)Coze, Dify=》低代码(可视化)

2)LangChain, LangGraph, Qwen-Agent, LlamaIndex=》高代码,工作中主流方式,Function Call + MCP

3)OpenClaw, Hermes, WorkBuddy,DeepAgents=》现代Agent框架(agent + skills) 或者是work agent,高代码 =》适合业务同学,主要是skill的开发。

在实现Agent时,有三个核心原则:

1. 保持智能体设计的简洁性:避免不必要的复杂性,专注于核心功能。

2. 优先考虑透明性:明确展示智能体的规划步骤,让用户清楚了解其决策过程。

3. 打造Function/MCP:打造工具,以及说明文档和测试,确保Agent与外部环境的交互。

三、Agent分类

1、反应式(Reactive)

这个也就是我们前面提到过的初识Agent React Agent。反应式架构是AI智能体设计中最简单直接的模式。在这种架构 中,一个大型语言模型(LLM)首先分析当前情况,确定下一 步要采取的行动。然后,在环境中执行该行动,产生观察结果 作为反馈。LLM处理这些观察结果,重新评估下一步行动,选 择另一个行动,并继续这个循环,直到任务完成。

反应式架构:快速决策的“直觉型”智能体

CASE:私募基金运作指引问答助手

之前我们做问答助手,是用Rag来完成的:RAG搭建本地知识库,RAG多模态知识库。今天这个示例不用Rag,使用LangChain框架,用工具的方式来完成。

Q:Rag和调用知识检索的工具这两方法的区别?

A:1.Rag的流程固定,基于Query和chunk进行相似度的匹配,结果可复现;Agent调用工具是灵活的,根据LLM的推理结果来选择

2.如果在检索过程中有新的需求,Rag没法给到支持,而Agent能灵活应对。

FUND_RULES_DB = [ { "id": "rule001", "category": "设立与募集", "question": "私募基金的合格投资者标准是什么?", "answer": "合格投资者是指具备相应风险识别能力和风险承担能力,投资于单只私募基金的金额不低于100万元且符合下列条件之一的单位和个人:\n1. 净资产不低于1000万元的单位\n2. 金融资产不低于300万元或者最近三年个人年均收入不低于50万元的个人" }, ...... } @tool def search_rules_by_keywords(keywords: str) -> str: """通过关键词搜索相关私募基金规则。输入应为相关关键词,多个关键词用逗号或空格分隔。""" ....... @tool def search_rules_by_category(category: str) -> str: """根据规则类别查询私募基金规则。输入应为类别名称,可选类别:设立与募集、监管规定。""" ....... @tool def answer_question(query: str) -> str: """在知识库中搜索并回答用户关于私募基金的问题。输入应为完整的用户问题。""" ....... def create_fund_qa_agent(): llm = ChatTongyi(model="qwen-plus", dashscope_api_key=DASHSCOPE_API_KEY) tools = [search_rules_by_keywords, search_rules_by_category, answer_question] system_prompt = """你是一个私募基金问答助手,专门回答关于私募基金规则和运作的问题。 你可以使用以下工具来查询信息: 1. search_rules_by_keywords: 通过关键词搜索相关规则 2. search_rules_by_category: 按类别查询规则(类别:设立与募集、监管规定) 3. answer_question: 直接搜索并回答问题 注意: 1. 如果知识库中没有相关信息,请明确告知用户"对不起,在我的知识库中没有关于[具体主题]的详细信息" 2. 如果你基于自己的知识提供补充信息,请用"根据我的经验"或"一般来说"等前缀明确标识 3. 回答要专业、简洁、准确""" agent = create_agent(llm, tools, system_prompt=system_prompt) return agent #--------------------------------------------------------- # user_input是用户输入的问题 response = fund_qa_agent.invoke({"messages": [HumanMessage(content=user_input)]})

在langChain中函数的定义如上所示,装饰器@tool注册函数。函数下面有三个引号包着的函数描述(python里叫注释),大模型读取这个函数描述帮助做推理(固定写法)。如上,传入三个工具search_rules_by_keywords, search_rules_by_category, answer_question,放入tools。

这里面如何调工具,怎么调用工具是LLM来推理决定的。因此选择工具是不确定的,如同一个问题,它有可能选择A工具执行,也可能选择B工具执行。因此,我们在添加工具时,尽量使工具的功能不要重叠,增强agent的稳定性。

2、深思熟虑(Deliberative)

这个类型,我理解为之前提到过的workflow型Agent。他是适合有长期价值的,多步复杂任务,

CASE:智能投研助手

这个智能体能够整合市场数据,进行多步骤分析和推理,生成投资观点和研究报告。 与反应式智能体不同,深思熟虑型智能体通过内部建模、推理和规划,制定最优行动方案,特别适合需要多步骤思考和长期目标优化的任务。

我们使用 LangGraph框架来完成。LangGraph是langChain家族的一员,可以做更丰富的链路编排(有点像扣子编程)。

主要步骤:感知阶段,建模阶段,推理阶段,决策阶段,报告阶段

Step 1,定义输出模型:各个阶段输出的内容

# 定义输出模型 class PerceptionOutput(BaseModel): """感知阶段输出的市场数据和信息""" market_overview: str = Field(..., description="市场概况和最新动态") key_indicators: Dict[str, str] = Field(..., description="关键经济和市场指标") recent_news: List[str] = Field(..., description="近期重要新闻") industry_trends: Dict[str, str] = Field(..., description="行业趋势分析") class ModelingOutput(BaseModel): """建模阶段输出的内部世界模型""" market_state: str = Field(..., description="当前市场状态评估") economic_cycle: str = Field(..., description="经济周期判断") risk_factors: List[str] = Field(..., description="主要风险因素") opportunity_areas: List[str] = Field(..., description="潜在机会领域") market_sentiment: str = Field(..., description="市场情绪分析") class ReasoningPlan(BaseModel): """推理阶段生成的候选分析方案""" plan_id: str = Field(..., description="方案ID") hypothesis: str = Field(..., description="投资假设") analysis_approach: str = Field(..., description="分析方法") expected_outcome: str = Field(..., description="预期结果") confidence_level: float = Field(..., description="置信度(0-1)") pros: List[str] = Field(..., description="方案优势") cons: List[str] = Field(..., description="方案劣势") class DecisionOutput(BaseModel): """决策阶段选择的最优投资观点""" selected_plan_id: str = Field(..., description="选中的方案ID") investment_thesis: str = Field(..., description="投资论点") supporting_evidence: List[str] = Field(..., description="支持证据") risk_assessment: str = Field(..., description="风险评估") recommendation: str = Field(..., description="投资建议") timeframe: str = Field(..., description="时间框架")

Step 2,定义状态:

这个状态里的变量,就是工作流中公共的一些状态变量

class ResearchAgentState(TypedDict): """研究智能体的状态""" # 输入 research_topic: str # 研究主题 industry_focus: str # 行业焦点 time_horizon: str # 时间范围(短期/中期/长期) # 处理状态 perception_data: Optional[Dict[str, Any]] # 感知阶段收集的数据 world_model: Optional[Dict[str, Any]] # 内部世界模型 reasoning_plans: Optional[List[Dict[str, Any]]] # 候选分析方案 selected_plan: Optional[Dict[str, Any]] # 选中的最优方案 # 输出 final_report: Optional[str] # 最终研究报告 # 控制流 current_phase: Literal["perception", "modeling", "reasoning", "decision", "report"] error: Optional[str] # 错误信息

Step 3,创建工作流:

def create_research_agent_workflow() -> StateGraph: """创建深思熟虑型研究智能体工作流图""" # 创建状态图 workflow = StateGraph(ResearchAgentState) #公共状态 # 添加节点 workflow.add_node("perception", perception) #perception是处理函数 workflow.add_node("modeling", modeling) workflow.add_node("reasoning", reasoning) workflow.add_node("decision", decision) workflow.add_node("report", report_generation) # 设置入口点 workflow.set_entry_point("perception") # 设置边和转换条件 workflow.add_edge("perception", "modeling") workflow.add_edge("modeling", "reasoning") workflow.add_edge("reasoning", "decision") workflow.add_edge("decision", "report") workflow.add_edge("report", END) # 编译工作流 return workflow.compile() #------------------------------------------------ agent = create_research_agent_workflow() result = agent.invoke(initial_state)

Step 4,具体实现:

上面工作流中add_node绑定的函数主要流程如下:

五步里每一步构建提示词,调用LLM进行推理,然后返回状态。最重要的是提示词,定义输入输出,输出的内容等。

# 准备提示 prompt = ChatPromptTemplate.from_template(PERCEPTION_PROMPT) # 构建输入 input_data = { "research_topic": state["research_topic"], "industry_focus": state["industry_focus"], "time_horizon": state["time_horizon"] } # 调用LLM chain = prompt | llm | JsonOutputParser() result = chain.invoke(input_data) # 更新状态 return { **state, "perception_data": result, "current_phase": "modeling" }

我们链路的每一步的输入都是前一步的输出,需要在input_data里面把前一步的结果传进去:

"perception_data": json.dumps(state["perception_data"], ensure_ascii=False, indent=2),

调用结果:

我们的整个流程如下:

3、混合式(Hybrid)

这种方式顾名思义的话是上面两种的结合。

到底选用反应式还是深思熟虑式,主要由中间的协调层来决定。

CASE:投顾AI助手

混合智能体采用三层架构设计:

协调层首先评估客户查询,确定:

• 查询类型:紧急型、信息型或分析型

• 处理模式:反应式或深思熟虑

这种意图识别有两种方法:1.关键词匹配 2.LLM推理

评估基于以下因素:

• 查询的复杂性和时效性

• 所需数据和计算资源

• 客户期望的响应时间

Step 1:工作流中公共的一些状态变量:

class WealthAdvisorState(TypedDict): """财富顾问智能体的状态""" # 输入 user_query: str # 用户问题 customer_profile: Optional[Dict[str, Any]] # 用户画像 # 处理状态 query_type: Optional[Literal["emergency", "informational", "analytical"]] # 查询类型 processing_mode: Optional[Literal["reactive", "deliberative"]] # 处理模式 market_data: Optional[Dict[str, Any]] # 市场数据 analysis_results: Optional[Dict[str, Any]] # 分析结果 # 消息历史(用于工具调用) messages: Annotated[List[BaseMessage], add_messages] # 输出 final_response: Optional[str] # 最终响应 # 控制流 current_phase: Optional[str] error: Optional[str]

Step 2:用LangGraph创建工作流

def create_wealth_advisor_workflow() -> StateGraph: """创建财富顾问混合智能体工作流""" workflow = StateGraph(WealthAdvisorState) # 添加节点 workflow.add_node("assess", assess_query) workflow.add_node("reactive_agent", reactive_agent) workflow.add_node("tools", tool_node) workflow.add_node("extract_response", extract_reactive_response) workflow.add_node("collect_data", collect_data) workflow.add_node("analyze", analyze_data) workflow.add_node("recommend", generate_recommendations) # 设置入口点 workflow.set_entry_point("assess") # 评估后的分支路由 workflow.add_conditional_edges( "assess", lambda x: "reactive_agent" if x.get("processing_mode") == "reactive" else "collect_data", { "reactive_agent": "reactive_agent", "collect_data": "collect_data" } ) # 反应式Agent的工具调用循环 workflow.add_conditional_edges( "reactive_agent", should_continue_tools, { "tools": "tools", "end": "extract_response" } ) # 工具执行后返回Agent继续处理 workflow.add_edge("tools", "reactive_agent") # 提取响应后结束 workflow.add_edge("extract_response", END) # 深思熟虑模式的流程 workflow.add_edge("collect_data", "analyze") workflow.add_edge("analyze", "recommend") workflow.add_edge("recommend", END) return workflow.compile() #--------------------------------------------------- agent = create_research_agent_workflow() result = agent.invoke(initial_state)

大家可以把节点列出来,在画板上画一下这个链路图,能更清晰一点。创建好agent以后把用户问题和用户画像传给agent(initial_state里包含)

Step 3:工作流开始,判断用哪一种agent

这里是让LLM进行判断

prompt = ChatPromptTemplate.from_template(ASSESSMENT_PROMPT) chain = prompt | llm | JsonOutputParser() result = chain.invoke({"user_query": state["user_query"]})

输出结果:

目前是走深思熟虑型这个分支,每一步都让大模型进行推理:

prompt = ChatPromptTemplate.from_template(ANALYSIS_PROMPT) chain = prompt | llm | JsonOutputParser() result = chain.invoke({ "user_query": state["user_query"], "customer_profile": json.dumps(state.get("customer_profile", {}), ensure_ascii=False, indent=2), "market_data": json.dumps(state.get("market_data", {}), ensure_ascii=False, indent=2) }) return { "analysis_results": result, "current_phase": "recommend" }

如果走反应式分支:

让大模型进行推理,这里是个循环,返回结果用should_continue_tools(Step 2创建工作流,返回“tools” 或者“end”)来判断是否需要调用工具(LLM返回内容里的tool_calls)

response = llm_with_tools.invoke(messages)

把大模型每次返回的结果更新到WealthAdvisorState的消息历史中,extract_reactive_response这个函数返回最后一条记录的结果。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询