目录一、什么是Agent二、Agent框架介绍三、Agent分类1、反应式(Reactive)CASE私募基金运作指引问答助手2、深思熟虑DeliberativeCASE智能投研助手3、混合式HybridCASE投顾AI助手在本篇的开头我们复习一下Agent的相关内容 初识Agent一、什么是Agent大模型是大脑Agent 是工程系统DeepSeek/Kimi 这类大模型本身更像“大脑”主要负责推理给输入产出判断或回答。但单个大脑不等于 AgentAgent 是把大模型放进一个工程框架里给它配上规划、工具、记忆等组件。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的开发框架分为以下三种1Coze, Dify》低代码可视化2LangChain, LangGraph, Qwen-Agent, LlamaIndex》高代码工作中主流方式Function Call MCP3OpenClaw, 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框架用工具的方式来完成。QRag和调用知识检索的工具这两方法的区别A1.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(modelqwen-plus, dashscope_api_keyDASHSCOPE_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_promptsystem_prompt) return agent #--------------------------------------------------------- # user_input是用户输入的问题 response fund_qa_agent.invoke({messages: [HumanMessage(contentuser_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_asciiFalse, indent2),调用结果我们的整个流程如下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_asciiFalse, indent2), market_data: json.dumps(state.get(market_data, {}), ensure_asciiFalse, indent2) }) return { analysis_results: result, current_phase: recommend }如果走反应式分支让大模型进行推理这里是个循环返回结果用should_continue_toolsStep 2创建工作流返回“tools” 或者“end”来判断是否需要调用工具LLM返回内容里的tool_callsresponse llm_with_tools.invoke(messages)把大模型每次返回的结果更新到WealthAdvisorState的消息历史中extract_reactive_response这个函数返回最后一条记录的结果。