尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

一个迷你版的聊天智能助手

一个迷你版的聊天智能助手 本文从 Agent、Workflow、Chatbot 三种形态的边界讲起拆解 Agent 的四大组成与运行模式再结合 Function Calling、MCP 和 LangChain 框架最后通过迷你聊天助手「小D」的实战演练带你从理论到代码完整落地一个 Agent 应用。文章目录一、Agent相关理论知识一What is Agent二Agent/Workflow/Chatbot三种形态的边界三Agent四大组成四函数调用Function Calling五MCPModel Context Protocol模型上下文协议二、 LangChain三、实战演练迷你版智能聊天助手小D一、Agent相关理论知识一What is Agent一些人定义Agent为完全自主的系统能够在较长时间内独立运行使用各种工具完成复杂任务也有人将Agent定义为遵循预定义工程的更具规范性的实现。一个常见的定义Agent LLM推理Tools行动能力Memory记忆Planning规划Agent不是一次性生成答案而是经过多轮思考-行动-观察的循环逐步接近最终答案二Agent/Workflow/Chatbot三种形态的边界在讨论 Agent 之前有必要先厘清它与 Workflow、Chatbot 之间的边界。三者都基于大语言模型但在流程控制权、自主程度和适用场景上存在明显差异。Chatbot是最常见也最基础的一种形态。它的特点是单轮或多轮对话每次回答都由用户输入直接触发模型根据上下文生成回复但不会主动规划下一步行动。典型实现包括客服机器人、知识问答助手等适合信息查询、简单问答等场景。Workflow介于 Chatbot 与 Agent 之间。它的特点是流程由开发者预先编排每一步做什么、调用什么工具、按什么顺序执行都是写死的。模型只在预设的节点上发挥作用例如先调用检索工具、再调用生成模型、最后做格式整理。典型实现包括 RAG 流水线、自动化报表生成等适合流程固定、步骤明确的业务场景。Agent则是三者中自主程度最高的形态。它的特点是流程由模型动态决定模型根据目标自主规划、调用工具、观察结果并调整下一步行动形成思考-行动-观察的循环直到完成任务。典型实现包括自主编程助手、多步骤研究助手等适合目标开放、路径不固定的复杂任务。形态流程由谁决定典型实现适合场景Chatbot用户输入直接触发模型即时回复客服机器人、知识问答助手信息查询、简单问答Workflow开发者预先编排流程固定RAG 流水线、自动化报表流程固定、步骤明确的业务Agent模型动态规划自主决策自主编程助手、多步骤研究助手目标开放、路径不固定的复杂任务怎么选择当任务有较强的不确定性需要根据环境反馈进行动态规划、工具选择和路径调整的时候适合用Agent如果任务流程和执行规则可以预先确定则优先采用Workflow。三Agent四大组成1.大语言模型LLMLarge Language Model2.规划模块PlanningAgent的运行模式1ReAct模式最核心的模式核心就是系统提示词需要包含模型角色、运行规则以及环境信息等。系统提示词和用户任务应该分开传给模型。ReAct模式可以根据中间结果调整计划适合探索性任务。但是Token消耗大每一步都需要先调用LLM。#ReAct实现示例 #核心就是tool装饰工具 create_agent(model,tools) invoke({messages:}) from langgraph.prebuilt import create_react_agent from langchain_core.tools import tool from langchain_openai import ChatOpenAI 定义工具用 tool装饰器 def search(query: str) - str: 搜索互联网信息 return search_function(query) # 你自己的实现 model ChatOpenAI( modeldeepseek-chat, api_keysk-xxx, base_urlhttps://api.deepseek.com/v1, ) agent create_react_agent(model, tools[search]) result agent.invoke( {messages: [{role: user, content: 找出DeepSeek-R1的训练成本并与GPT-4对比}]} ) print(result[messages][-1].content)2Plan-and-Execute先计划后执行只需要调用一次LLM规划【高效】可以并行执行多个任务。但是不灵活难以根据中间结果调整。#Plan-and-Execute实现伪代码 from langchain.agents import Plan, Execute 第一步规划 planner create_planner(llm) plan planner.plan(分析2024年AI Agent市场趋势) 输出的计划 Task 1: 搜索2024年AI Agent市场报告 Task 2: 提取市场规模数据 Task 3: 识别主要参与者 Task 4: 总结趋势和预测 第二步执行 executor create_executor(tools) results executor.execute(plan) # 按计划依次执行 #目前LangChain官方推荐使用LangGraph来实现Plan-and-Execute 核心架构通常包含三个部分 Planner规划器 Executor执行器 Controller控制器/重规划期 建议参考 LangChain 官方文档中的 Plan-and-Execute Agent 教程或者直接使用 LangGraph 框架。 #最新写法 from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, List import operator 1. 定义状态 class PlanExecute(TypedDict): input: str plan: List[str] past_steps: Annotated[List[tuple], operator.add] response: str 2. 定义 Planner 节点 (生成计划) def plan_step(state: PlanExecute): # 这里使用一个 LLM Chain 来生成计划 plan planner_chain.invoke({messages: [(user, state[input])]}) return {plan: plan.steps} 3. 定义 Executor 节点 (执行当前步骤) def execute_step(state: PlanExecute): task state[plan][0] # 使用 Agent 执行任务 agent_response executor_agent.invoke({input: task}) return { past_steps: [(task, agent_response[output])], plan: state[plan][1:] # 移除已执行的任务 } 4. 定义 Replan 节点 (决定是否继续还是结束) def replan_step(state: PlanExecute): # 检查计划是否为空或者是否得到最终答案 if not state[plan]: return {response: 任务完成} else: # 根据需要可以在这里重新规划 return {plan: state[plan]} 5. 构建图 workflow StateGraph(PlanExecute) workflow.add_node(planner, plan_step) workflow.add_node(executor, execute_step) workflow.add_node(replan, replan_step) workflow.set_entry_point(planner) workflow.add_edge(planner, executor) workflow.add_edge(executor, replan) workflow.add_conditional_edges( replan, lambda x: executor if x[plan] else END, # 如果还有计划则继续否则结束 ) app workflow.compile()怎么选择探索性任务不知道中间会碰到什么-- ReAct流程明确的任务步骤固定-- Plan-and-Execute复杂混合任务 -- 两者结合3Chain of Thought(CoT) 纯推理只有思考没有行动适合纯逻辑推理问题Token消耗少3.记忆模块Memorya短期记忆Short-term Memory保存当前任务的上下文存储内容当前对话历史、中间步骤的结果、临时变量和状态会有Token限制并且每次调用都会把整个历史都发送给LLM --使用摘要记忆b长期记忆Long-term Memory存储可复用的知识和经验存储内容用户的个人信息和偏好、历史任务的成功经验、领域知识库、工具使用的最佳实践技术实现使用向量数据库4.工具集Tools工具是Agent与外部世界交互的桥梁让Agent突破大语言模型自身的局限获得实时信息获取、外部系统操作和物理世界控制等能力。没有工具的Agent只能基于训练数据做推理而有了工具Agent才能真正动手做事。1工具的主要类型信息检索类搜索引擎、知识库查询、数据库检索、API数据获取等帮助Agent获取实时或领域特定信息。代码执行类Python解释器、代码沙箱、SQL执行器等让Agent能够运行代码、处理数据、进行计算。文件操作类文件读写、格式转换、数据导入导出等支持Agent处理本地或云端文件。外部服务类邮件发送、日历管理、支付接口、消息通知等让Agent能够调用第三方服务完成实际业务。专业领域类针对特定场景定制的工具如数学计算引擎、图像处理、语音合成、地图导航等。2工具的定义与注册在LangChain/LangGraph等框架中工具通常通过tool装饰器定义并注册到Agent中。每个工具需要包含函数名称、功能描述供LLM理解何时调用、参数定义JSON Schema格式以及具体的执行逻辑。from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的实时天气信息 Args: city: 城市名称如北京 Returns: 该城市的天气情况描述 # 调用天气API获取数据 result weather_api.query(city) return f{city}当前天气{result[condition]}温度{result[temperature]}℃3工具调用的工作流程Agent使用工具的过程遵循思考-行动-观察的循环LLM根据用户任务和当前上下文判断需要调用哪个工具生成包含工具名称和参数的结构化调用指令系统执行工具并返回结果LLM观察结果决定是继续调用其他工具还是生成最终答案。4工具选择的策略当Agent拥有多个工具时需要合理选择。常见策略包括根据任务类型匹配工具描述、通过ReAct模式动态决策、使用路由机制按规则分发。工具描述写得越清晰LLM选择工具的准确率就越高。四函数调用Function CallingAgent 能查天气、能搜索信息、能操作数据库这些能力是怎么实现的答案就是Function Calling函数调用。Function Calling四个流程步骤用户提问 -- LLM决策LLM确定函数调用参数 -- 应用执行实际调用API -- LLM生成根据结果生成最终答案第一步定义函数。开发者预先告诉LLM【你手边有哪些工具可以用】用JSON格式描述每个函数的名字、功能说明和参数。比如你告诉它有一个get_weather函数接收一个城市名参数返回天气信息。{ tools: [ { type: function, function: { name: get_weather, description: 获取指定城市的实时天气, parameters: { type: object, properties: { city: { type: string, description: 城市名称比如上海 } }, required: [city] } } } ] }第二步模型判断。用户提问后LLM分析用户的意图自己判断【要回答这个问题我需要调用哪个函数】。如果用户问【上海今天天气怎么样】LLM会决定调用get_weather并生成参数{city:上海}{ tool_calls: [ { type: function, function: { name: get_weather, arguments: {\city\: \上海\} } } ] }第三步执行函数。LLM自己并不执行函数。它只是输出了【我想调用这个函数参数是这些】的结构化指令。真正执行函数的是你的应用程序。你的代码拿到了LLM返回的调用指令后解析出了city上海去实际调用天气API拿到结果比如22度多云。第四步生成回答。你的代码把拿到的真实温度数据再次发给LLM。LLM这次有了客观的数据支撑就会用非常自然的人类语言回答你今天上海天气是多云气温大约22摄氏度。Function Calling是一次性的【单步调用】LLM判断需要调用一个函数调用完就结束了。而Agent是【循环调用】Agent在一个循环中反复调用Function Calling每次调用后观察结果再决定下一步要不要继续调用其它函数。每个LLM厂商的Function Calling格式不一样五MCPModel Context Protocol模型上下文协议MCP提供了一个统一的标准让任何AI应用都能够用一种方式链接任何外部工具和数据源。MCP架构首先是MCP Host宿主就是你使用的 AI 应用比如 Claude Desktop、Cursor 编辑器、你自己开发的 Agent 应用。它是整个交互的发起方。- 然后是MCP Client客户端它住在 Host 里面负责跟 MCP Server 通信。你可以把它理解为翻译官Host 想要什么能力Client 就去跟对应的 Server 沟通。- 最后是MCP Server服务端它负责对外暴露具体的工具能力和数据资源。比如有一个 GitHub MCP Server它能提供搜索代码创建 Issue查看 PR等工具。一个 Slack MCP Server 能提供发送消息搜索频道等工具。整个流程就是用户在 AI 应用中提问 → AI 应用Host通过 MCP Client 发现有哪些可用工具 → AI 决定调用某个工具 → MCP Client 向对应的 MCP Server 发送请求 → Server 执行操作返回结果 → AI 基于结果生成回答。二、 LangChainLangChain是一个用于开发LLM应用的框架。LangChain四大核心Model Prompt Tool Agent六大模块1.Models模型LangChain 对各类模型做了统一抽象主要分三类LLM纯文本补全模型如早期的 GPT-3Chat Model对话模型如 GPT-4、Claude、通义千问输入输出是消息列表Embedding Model把文本转成向量用于检索、相似度计算2.Prompts提示词负责把用户输入、上下文、历史消息组装成模型能理解的格式。PromptTemplate字符串模板ChatPromptTemplate消息列表模板system / human / aiFew-shot 示例、输出解析指令也常在这里定义3.Chains链把多个组件串起来形成一条处理流水线。4.Memory记忆让对话具备“记忆”能力保存历史消息。常见类型ConversationBufferMemory全量保存ConversationSummaryMemory摘要压缩ConversationBufferWindowMemory只保留最近 N 轮5.Retrieval检索让模型能够访问外部知识6.Agents智能体让 LLM 自主决定调用哪个工具、按什么顺序执行。Agent LLM Tools 循环决策。Tools搜索、计算器、API 调用等Agent Executor驱动“思考→行动→观察”的循环三、实战演练迷你版智能聊天助手小D详细代码看Stella0412/Mini_smart_assistant: 一个迷你的初级聊天智能体项目结构main.py 负责「编排」tools/ 负责「能力」。每新增一个能力只需在 tools/ 下加一个带 tool 装饰器的函数再注册到 create_agent 的 tools 列表即可——这就是 Agent 应用典型的插件式扩展结构。Mini_smart_assistant-main/ ├── main.py # 入口创建模型 / 记忆 / Agent跑命令行对话循环 ├── requirements.txt # 依赖清单 ├── .gitignore # 忽略 .env、__pycache__、uploads/ 等 ├── 一个迷你版的聊天智能助手.md # 环境搭建笔记 ├── tools/ # 工具包每个文件 一类能力 │ ├── __init__.py # 空文件标记为 Python 包 │ ├── calculator.py # 数学计算工具 │ ├── weather.py # 天气查询工具Tavily 搜索 │ ├── translator.py # 翻译工具Google 翻译 │ ├── knowledge.py # 知识库查询工具字典匹配 │ └── readfile.py # 文件读取工具含路径安全校验 └── uploads/ └── test.txt # 示例文件用于演示 read_user_file1.环境搭建需要在项目中创建虚拟环境#创建一个虚拟环境 langchain_env是虚拟环境名称 python -m venv langchain_env #激活虚拟环境 langchain_env\Scripts\activate #导入Deepseek pip install -U langchain-deepseek #环境变量管理 pip install python-dotenv #Agent运行时 pip install langgraph #tavily-python是一个强大的 AI 搜索 API 客户端用于获取实时的网络搜索结果可以在官网注册后获取 API。 pip install requests tavily-python2.配置API KEY在根目录下创建.env文件存储API KEYDEEPSEEK_API_KEYYOUR_DEEPSEEK_API_KEY TAVILY_API_KEYYOUR_TAVILY_API_KEY3.编写工具Tool以get_weather工具为例from langchain_core.tools import tool from langchain_tavily import TavilySearch tool def get_weather(city:str) - str: 查询指定城市的实时天气情况。 Args: city: 城市名称例如北京、上海 # 创建 Tavily 搜索工具默认自动读取环境变量 TAVILY_API_KEY search TavilySearch(max_results3, topicgeneral) #进行错误处理返回字符串而不是抛异常 try: results search.invoke({query: f{city} 今天实时天气预报 气温}) except Exception as e: return f错误:调用Tavily搜索时出现问题 - {e} if not results.get(results): return f未找到关于{city}的天气信息请确认城市名称是否正确 # 整理搜索结果为易读的自然语言摘要 lines [f【{city}天气查询结果】] for i, item in enumerate(results[results], 1): title item.get(title, ) content (item.get(content, ) or ).strip()[:200] url item.get(url, ) lines.append(f\n{i}. {title}) if content: lines.append(f {content}) if url: lines.append(f 来源: {url}) return \n.join(lines)一个Tool是由三部分组成的tool装饰器、函数签名参数类型注解、docstring要素作用tool装饰器把普通函数升级为 LangChain 可识别的工具对象函数签名告诉 LLM 这个工具需要什么参数、类型是什么docstringLLM 靠它判断什么时候该调用这个工具返回值str工具执行结果回传给 LLMLLM 看不到你的函数体只能看到函数名、参数、docstring。所以 docstring 写得好不好直接决定模型会不会在正确的时机调用它。为什么把错误作为字符串返回而不是raise【raise就是抛出异常将函数中断执行】因为工具的输出会作为观察结果回传给 LLM。如果你返回一段可读的错误信息LLM 就有机会换一个城市名重试告诉用户暂时查不到改用其他工具如果直接抛异常整个 Agent 链路可能直接中断。错误也是给 LLM 的信息是写 Tool 的一条重要心法。其它工具写法类似4.主函数import os from dotenv import load_dotenv from langchain.agents import create_agent from langchain_deepseek import ChatDeepSeek from langgraph.checkpoint.memory import MemorySaver #导入工具 from tools.translator import translate from tools.weather import get_weather from tools.knowledge import search_knowledge from tools.calculator import calculator from tools.readfile import read_user_file #加载环境变量 load_dotenv() #系统提示词 SYSTEM_PROMPT 你是小D一个全能智能助手 你可以帮用户 查询天气查询中国主要城市的天气 数学运算进行各种数学运算 文本翻译将中文翻译成其他语言 搜索知识给用户提供知识科普 读取文件使用 read_user_file 工具读取用户提供的文件 文件读取规则 1. 当用户要求总结、分析、查看或读取文件时使用 read_user_file 工具。 2. 用户需要提供文件路径例如 uploads/test.txt。 3. 如果用户没有提供文件路径先询问用户文件保存在哪里。 4. 只能读取 uploads 目录下的文件。 5. 读取文件后可以根据文件内容进行总结、提取信息和回答问题。 6. 不要假设没有读取到的文件内容。 工作原则 1.先理解用户意图选择合适的工具 2.如果不确定可以询问用户 3.回答简洁明了有帮助 def create_assistant(): 创建智能助手 #初始化模型 llm ChatDeepSeek( model deepseek-flash, temperature0.7, api_keyos.environ.get(DEEPSEEK_API_KEY) ) #创建记忆 checkpoint MemorySaver() #创建Agent agent create_agent( modelllm, tools[calculator,search_knowledge,get_weather,translate,read_user_file], system_promptSYSTEM_PROMPT, checkpointercheckpoint ) return agent def main(): 主函数 print(*50) print( 小D - 全能智能助手v1.0) print(*50) print(输入 quit 退出\n) agent create_assistant() config {configurable:{thread_id:main}} while True: user_input input(你: ).strip() if user_input.lower() in [quit, 退出, exit]: print(再见) break if not user_input: continue try: result agent.invoke( {messages: [{role: user, content: user_input}]}, config ) response result[messages][-1].content print(f小D: {response}\n) except Exception as e: print(f出错了: {str(e)}\n) if __name__ __main__: main()
返回列表