1. 项目概述一个“会干活”的AI伙伴最近在折腾AI应用落地的朋友估计都绕不开一个核心痛点如何让大语言模型从一个“博学的聊天对象”真正变成一个能帮你“干脏活累活”的可靠伙伴这就是“WorkBuddy”这个项目想解决的根本问题。它不是一个简单的聊天机器人而是一个旨在深度理解你的身份、习惯和任务上下文并主动替你执行具体操作的智能体。简单来说传统的AI助手你问它“今天天气怎么样”它会告诉你天气。但WorkBuddy的思路是当你早上说“今天有什么安排”时它不仅能基于日历告诉你上午十点有会还能根据你“总是提前15分钟到会议室”的习惯在9:45提醒你动身甚至自动帮你把会议资料从云盘下载到本地并打开相关的演示文稿。它的目标是从被动应答的“我是谁”身份识别与记忆跃迁到主动服务的“帮我干活”任务理解与自动化执行。这个项目适合所有对AI应用开发、智能体构建以及自动化工作流感兴趣的朋友无论是想提升个人效率的极客还是希望为企业内部打造智能助手的开发者。接下来我会拆解实现这样一个“WorkBuddy”的核心设计思路、关键技术栈并分享从零搭建原型过程中那些必须注意的“坑”。2. 核心架构设计让AI拥有“手”和“记忆”要实现从认知到执行的跨越系统的架构设计是关键。你不能只靠一个语言模型包打天下它需要一套组合拳。2.1 三层核心组件解析一个完整的WorkBuddy系统可以抽象为三层感知与理解层、决策与规划层、执行与反馈层。感知与理解层这是“我是谁”的基础。它的核心是用户画像与上下文管理。你需要一个向量数据库来存储和检索用户的历史对话、行为偏好、个人资料。例如用户提到过“我习惯用Markdown写周报”这句话需要被解析、嵌入成向量存起来。下次用户说“写周报”时系统不仅能检索到“周报”这个任务还能关联到“Markdown格式”这个偏好。这里常用的工具有ChromaDB、Pinecone或者Weaviate。关键在于设计好数据的嵌入和检索策略比如将用户信息、对话历史、任务结果分开存储通过元数据过滤提高检索精度。决策与规划层这是大脑。大语言模型在这里扮演核心角色但它不是直接输出答案而是输出一个可执行的计划。当用户说“帮我整理一下上周的项目资料”模型需要拆解任务1. 确定时间范围上周一至周日2. 定位资料存储位置可能是Google Drive的“ProjectX”文件夹3. 定义“整理”的具体操作按日期重命名文件、生成摘要目录4. 识别所需工具调用云盘API、调用文件处理API。这个过程被称为“任务分解”或“思维链规划”。我们通常使用像GPT-4、Claude 3或者开源的Llama 3、DeepSeek等具有较强推理能力的模型并通过精心设计的提示词来引导其进行结构化输出。执行与反馈层这是“手”。AI自己不能操作电脑它需要通过工具调用来实现。你需要为模型提供一套“工具集”比如read_file、search_web、send_email、execute_shell_command需极其谨慎等。模型根据规划层的输出决定调用哪个工具并生成正确的调用参数。执行后工具返回结果成功或失败附带数据这个结果会反馈给模型模型根据结果决定下一步是继续执行还是向用户请求澄清。这一层的实现框架ReAct和LangChain是很好的起点它们提供了将模型、工具、记忆连接起来的标准化模式。注意工具调用是安全风险最高的环节。绝对禁止开放高危工具如无条件执行任意Shell命令、直接访问数据库写操作给模型。必须实施严格的工具权限管控和参数校验。例如文件删除工具只能删除特定临时目录下的文件邮件发送工具需要二次确认或限制收件人域名。2.2 状态管理与会话持久化WorkBuddy需要记住跨会话的信息。这不仅仅是保存聊天记录那么简单而是维护一个动态的会话状态。这个状态包括当前正在执行的多步骤任务进度、已收集到的用户输入参数、临时生成的数据等。例如用户说“订一张明天去上海的机票”WorkBuddy进入订票流程状态变为awaiting_departure_time。你回复“下午出发”状态更新并接着问“您偏好哪个航空公司”。实现上可以将整个会话状态一个复杂的Python字典或Pydantic模型序列化后与用户ID关联存储到数据库如Redis用于快速会话缓存PostgreSQL用于长期持久化。每次交互开始前先加载状态这样模型就知道“对话进行到哪一步了”。LangChain的ConversationChain或自定义的StateGraph在LangGraph中是管理这类状态的强大工具。3. 关键技术实现与工具链选型理论讲完我们落到具体的代码和工具上。我会以构建一个简单的、能处理“文件整理”和“信息查询”任务的WorkBuddy原型为例。3.1 搭建基础环境与智能体核心首先我们选择LangChain作为智能体框架因为它生态丰富抽象得当。假设我们使用OpenAI的GPT-4作为核心模型。# 环境准备 pip install langchain langchain-openai langchain-community chromadb python-dotenv# core_agent.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain.memory import ConversationBufferMemory from langchain.agents import AgentExecutor, create_react_agent from langchain.tools import Tool from langchain.prompts import PromptTemplate from langchain_community.tools import DuckDuckGoSearchRun # 假设我们自定义了一些工具后面会定义 from custom_tools import FileSearchTool, SummarizeDocumentTool load_dotenv() # 1. 初始化大模型 llm ChatOpenAI( modelgpt-4-turbo-preview, temperature0.1, # 低温度保证任务执行的稳定性 api_keyos.getenv(OPENAI_API_KEY) ) # 2. 初始化记忆用于存储对话历史 memory ConversationBufferMemory( memory_keychat_history, return_messagesTrue, output_keyoutput ) # 3. 定义工具集 search DuckDuckGoSearchRun() file_search FileSearchTool() summarizer SummarizeDocumentTool() tools [ Tool( nameWeb Search, funcsearch.run, descriptionUseful for when you need to answer questions about current events or general knowledge. Input should be a clear search query. ), Tool( nameFile Search, funcfile_search.run, descriptionUseful for searching and retrieving information from the users document repository. Input should be keywords related to the file content. ), Tool( nameDocument Summarizer, funcsummarizer.run, descriptionUseful for summarizing long documents. Input should be the path to a text file or the text content itself. ) ] # 4. 创建ReAct智能体 prompt PromptTemplate.from_template( You are WorkBuddy, a helpful assistant that can perform tasks. You have access to the following tools: {tools} Use the following format: Question: the input question you must answer Thought: you should always think about what to do Action: the action to take, should be one of [{tool_names}] Action Input: the input to the action Observation: the result of the action ... (this Thought/Action/Action Input/Observation can repeat N times) Thought: I now know the final answer Final Answer: the final answer to the original input Begin! Previous conversation history: {chat_history} Question: {input} Thought:{agent_scratchpad} ) agent create_react_agent(llm, tools, prompt) agent_executor AgentExecutor( agentagent, toolstools, memorymemory, verboseTrue, # 开发时打开查看思考过程 handle_parsing_errorsTrue, max_iterations5 # 防止死循环 ) # 运行示例 if __name__ __main__: response agent_executor.invoke({input: 帮我找一下上个月关于市场分析的文档并总结核心观点。}) print(response[output])这个代码搭建了一个最基本的、具备思考Thought和行动Action能力的智能体。它可以通过搜索工具找文件然后用总结工具生成摘要。3.2 实现自定义工具以文件操作为例上面用到的FileSearchTool和SummarizeDocumentTool需要我们自己实现。这里展示一个连接简单本地文件系统的搜索工具。# custom_tools.py import os from pathlib import Path from typing import Type from langchain.tools import BaseTool from pydantic import BaseModel, Field from langchain_community.document_loaders import TextLoader, PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings class FileSearchInput(BaseModel): query: str Field(descriptionKeywords to search for within documents) directory: str Field(default., descriptionDirectory to search in) class FileSearchTool(BaseTool): name file_search description Searches for content within text and PDF files in a specified directory. args_schema: Type[BaseModel] FileSearchInput return_direct: bool False # 结果返回给Agent继续处理 vector_store None def _setup_vector_store(self, directory: str): 初始化或加载指定目录的文档向量库 persist_path f./chroma_db/{os.path.basename(directory)} embeddings OpenAIEmbeddings() if os.path.exists(persist_path): # 加载已有库 self.vector_store Chroma( persist_directorypersist_path, embedding_functionembeddings ) else: # 构建新库 documents [] for ext in [*.txt, *.md, *.pdf]: for file_path in Path(directory).rglob(ext): try: if file_path.suffix .pdf: loader PyPDFLoader(str(file_path)) else: loader TextLoader(str(file_path), encodingutf-8) loaded_docs loader.load() # 为每个文档添加来源元数据 for doc in loaded_docs: doc.metadata[source] str(file_path) doc.metadata[filename] file_path.name documents.extend(loaded_docs) except Exception as e: print(fError loading {file_path}: {e}) if documents: # 分割文本便于检索 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200 ) split_docs text_splitter.split_documents(documents) # 创建向量存储 self.vector_store Chroma.from_documents( documentssplit_docs, embeddingembeddings, persist_directorypersist_path ) self.vector_store.persist() else: self.vector_store None def _run(self, query: str, directory: str .) - str: if not query: return Please provide a search query. # 懒加载向量库 if self.vector_store is None: self._setup_vector_store(directory) if self.vector_store is None: return fNo indexable text or PDF files found in {directory}. # 执行相似性搜索 try: results self.vector_store.similarity_search_with_relevance_scores(query, k3) if not results: return No relevant documents found for your query. response Here are the most relevant documents I found:\n for i, (doc, score) in enumerate(results): response f\n{i1}. **{doc.metadata.get(filename, Unknown)}** (Relevance: {score:.2f})\n response f Source: {doc.metadata.get(source)}\n response f Snippet: {doc.page_content[:200]}...\n return response except Exception as e: return fAn error occurred during search: {str(e)} async def _arun(self, query: str, directory: str .): raise NotImplementedError(Async operation not supported yet.)这个工具做了几件关键事1. 它接受查询和目录参数2. 首次使用时会加载目录下的文档并构建向量索引这是一个耗时的初始化过程实际产品中需要异步或后台处理3. 执行语义搜索返回最相关的文档片段及其来源。这比简单文件名搜索强大得多因为它理解内容。3.3 用户画像与长期记忆的实现让WorkBuddy记住“我是谁”我们需要一个用户画像系统。一个简单的实现可以基于向量数据库存储用户相关的所有信息片段。# user_profile.py from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain.schema import Document from datetime import datetime import json class UserProfileManager: def __init__(self, user_id: str): self.user_id user_id self.profile_store_path f./user_profiles/{user_id} self.embeddings OpenAIEmbeddings() self._init_store() def _init_store(self): 初始化或加载用户的向量化记忆库 try: self.vector_store Chroma( persist_directoryself.profile_store_path, embedding_functionself.embeddings ) except: # 首次创建 self.vector_store Chroma.from_documents( documents[], embeddingself.embeddings, persist_directoryself.profile_store_path ) def add_memory(self, memory_text: str, memory_type: str conversation, metadata: dict None): 添加一段记忆到用户画像 if metadata is None: metadata {} metadata.update({ user_id: self.user_id, type: memory_type, timestamp: datetime.now().isoformat() }) doc Document(page_contentmemory_text, metadatametadata) self.vector_store.add_documents([doc]) self.vector_store.persist() def query_memory(self, query: str, memory_type: str None, k: int 3): 查询与当前问题相关的用户记忆 filter_dict {user_id: self.user_id} if memory_type: filter_dict[type] memory_type results self.vector_store.similarity_search_with_relevance_scores( query, kk, filterfilter_dict ) return results def get_user_context(self, current_query: str) - str: 获取与当前查询相关的用户上下文用于拼接到提示词中 relevant_memories self.query_memory(current_query) if not relevant_memories: return No specific user context found. context_str Relevant user context from past interactions:\n for i, (doc, score) in enumerate(relevant_memories): context_str f- {doc.page_content} (from {doc.metadata.get(type, unknown)})\n return context_str # 使用示例 if __name__ __main__: user_manager UserProfileManager(user_123) # 添加一些记忆 user_manager.add_memory( The user prefers to receive summaries in bullet points., memory_typepreference ) user_manager.add_memory( The user is working on a project called Phoenix related to market analysis., memory_typeproject ) # 查询 context user_manager.get_user_context(How should I format the report?) print(context)这样每次用户交互时我们可以先调用get_user_context获取相关历史信息并将其作为系统提示词的一部分让模型在了解用户背景的情况下进行回复和规划。4. 提示词工程引导AI正确规划与执行智能体的“智商”很大程度上取决于提示词。对于WorkBuddy我们需要设计多阶段的提示词。4.1 系统提示词设计系统提示词定义了AI的角色、能力和行为规范。这是最核心的部分。你是一个名为WorkBuddy的智能助手。你的核心目标是理解用户需求并主动、准确地完成任务。 **你的身份与原则** 1. 你是用户的专属助手熟知用户的偏好和历史上下文会提供给你。 2. 你拥有工具使用能力。在回答问题或执行任务时应优先考虑是否可以通过调用工具更高效、准确地完成。 3. 对于复杂任务你必须先制定分步计划然后逐步执行。每次只执行一步观察结果后再决定下一步。 4. 如果工具执行失败或结果不明确不要猜测。向用户澄清或尝试替代方案。 5. 始终以帮助用户完成具体工作为目标回答应简洁、务实避免冗长的理论阐述。 **可用的工具** {tools} **当前用户上下文** {user_context} **对话历史** {chat_history} 请严格按照以下格式输出 Thought: 分析用户请求结合上下文和历史。决定是否需要使用工具以及使用哪个工具。 Action: 需要使用的工具名如果不使用工具则为 None。 Action Input: 工具的输入参数如果不使用工具则为 None。 Observation: 工具返回的结果第一步先留空。 ...重复 Thought/Action/Action Input/Observation 直到任务完成或无需更多工具 Thought: 我现在有足够信息给出最终答案。 Final Answer: 对用户的最终回复汇总结果或确认任务完成。这个提示词明确了ReAct格式并强调了规划、工具优先和基于上下文的行动。4.2 任务分解与参数提取提示词对于模糊的用户指令我们需要一个专门的步骤来澄清和分解。这可以通过一个独立的“规划器”LLM调用来实现。# planner_prompt.py from langchain.prompts import ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser from pydantic import BaseModel, Field from typing import List class Subtask(BaseModel): description: str Field(description对该子任务的清晰描述) tool: str Field(description完成此任务所需的工具名称如 FileSearchTool, Web Search, 或 None) parameters: dict Field(default_factorydict, description调用工具所需的参数) class TaskPlan(BaseModel): main_goal: str Field(description用户的主要目标) subtasks: List[Subtask] Field(description为达成目标所需的子任务序列) needs_clarification: bool Field(description是否需要向用户询问更多信息) clarification_question: str Field(default, description如果需要澄清要问什么问题) planner_prompt ChatPromptTemplate.from_messages([ (system, 你是一个任务规划专家。请将用户的模糊请求分解为具体、可执行的子任务步骤。 考虑可用的工具{tool_descriptions}。 如果用户请求缺少必要信息如时间、地点、具体文件名请标记需要澄清并生成一个清晰的问题。 输出必须是有效的JSON格式。), (human, 用户请求{user_input}\n当前已知用户上下文{user_context}) ]) def create_plan(user_input: str, user_context: str, llm): parser JsonOutputParser(pydantic_objectTaskPlan) full_prompt planner_prompt.format_parts( user_inputuser_input, user_contextuser_context, tool_descriptions, .join([f{t.name}: {t.description} for t in tools]) # 假设tools已定义 ) message full_prompt.to_messages() response llm.invoke(message) plan_dict parser.parse(response.content) return TaskPlan(**plan_dict)这个规划器会先于主智能体运行。如果needs_clarification为TrueWorkBuddy会先问用户那个问题。否则它就获得了一个清晰的子任务列表可以按顺序执行或交由智能体动态调整。5. 安全、隐私与工程化考量将AI智能体投入实际使用尤其是处理用户文件和数据时安全和隐私是生命线。5.1 工具执行沙箱与权限控制绝对不能允许AI直接、无限制地调用系统命令或访问所有文件。必须实现一个安全层。# security_layer.py import subprocess import os from pathlib import Path class SafeCommandTool: 一个安全的、受限的Shell命令执行工具 allowed_commands { list_dir: {cmd: [ls, -la], cwd: None}, current_date: {cmd: [date], cwd: None}, # 可以定义更多白名单命令 } allowed_directories [/tmp/workbuddy, os.path.expanduser(~/workbuddy_safe)] staticmethod def run(command_name: str, args: list None, working_dir: str None): if command_name not in SafeCommandTool.allowed_commands: return fError: Command {command_name} is not in the allowed list. config SafeCommandTool.allowed_commands[command_name].copy() cmd_list config[cmd] if args: cmd_list.extend(args) # 限制工作目录 safe_cwd working_dir if working_dir in SafeCommandTool.allowed_directories else config.get(cwd) if safe_cwd: Path(safe_cwd).mkdir(parentsTrue, exist_okTrue) try: result subprocess.run( cmd_list, cwdsafe_cwd, capture_outputTrue, textTrue, timeout10 ) if result.returncode 0: return result.stdout else: return fCommand failed with error: {result.stderr} except subprocess.TimeoutExpired: return Error: Command timed out. except Exception as e: return fError executing command: {str(e)}这个工具只允许执行预定义的白名单命令并在指定目录下运行。对于文件操作同样需要路径白名单或沙箱机制。5.2 数据隐私与匿名化处理如果WorkBuddy需要处理包含个人身份信息的数据必须在发送给外部AI API如OpenAI前进行匿名化。# privacy_filter.py import re class PrivacyFilter: patterns { email: r\b[A-Za-z0-9._%-][A-Za-z0-9.-]\.[A-Z|a-z]{2,}\b, phone_cn: r\b1[3-9]\d{9}\b, # 简单中国手机号匹配 id_card: r\b[1-9]\d{5}(18|19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[0-9Xx]\b # 简单身份证号匹配 } staticmethod def anonymize_text(text: str): anonymized text for ptype, pattern in PrivacyFilter.patterns.items(): anonymized re.sub(pattern, f[{ptype}_redacted], anonymized) return anonymized staticmethod def should_send_to_external_api(text: str, user_consent: bool): 决定文本是否可发送外部API if not user_consent: return False, No user consent for external API. # 检查是否包含高敏感信息可自定义规则 high_sensitivity_pattern r(password|token|key|secret)\s*[:]\s*\S if re.search(high_sensitivity_pattern, text, re.IGNORECASE): return False, Text contains high-sensitivity information. # 匿名化后发送 safe_text PrivacyFilter.anonymize_text(text) return True, safe_text在调用LLM API前先通过should_send_to_external_api检查并处理文本。同时所有本地存储的用户数据如向量库应进行加密。5.3 错误处理与鲁棒性增强智能体在复杂环境中必然会出错。健壮的错误处理机制必不可少。# error_handling.py from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import openai class RobustAgentExecutor: def __init__(self, agent_executor): self.agent agent_executor retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10), retryretry_if_exception_type((openai.APITimeoutError, openai.APIConnectionError)) ) def invoke_with_retry(self, input_dict): 带重试的调用主要处理网络或API瞬时错误 try: return self.agent.invoke(input_dict) except Exception as e: # 解析智能体特定的错误如工具调用失败、输出解析错误 error_msg str(e) if Could not parse LLM output in error_msg: # 模型输出不符合格式可能是复杂问题导致混乱 return { output: I encountered an issue while planning the steps. Could you please rephrase your request or break it down into simpler tasks? } elif Tool not found in error_msg: return { output: I tried to use a tool that isnt available. Let me try a different approach. } else: # 其他未预见的错误记录并返回友好信息 print(fUnhandled agent error: {error_msg}) return { output: An unexpected error occurred. Please try again or contact support if the problem persists. }此外要为每个工具调用设置超时并为智能体的“思考循环”设置最大迭代次数防止陷入死循环消耗资源。6. 部署与迭代从原型到可用产品让WorkBuddy真正可用需要把它包装成一个服务。6.1 后端API服务搭建使用FastAPI可以快速构建一个提供Web服务的后端。# main.py (FastAPI 后端) from fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel from typing import Optional import uuid from agent_core import get_agent_for_user # 假设这是封装好的智能体获取函数 from user_profile import UserProfileManager app FastAPI(titleWorkBuddy API) # 简单的内存会话存储生产环境用Redis user_sessions {} class UserRequest(BaseModel): message: str session_id: Optional[str] None class AgentResponse(BaseModel): reply: str session_id: str needs_follow_up: bool False app.post(/chat, response_modelAgentResponse) async def chat_with_buddy(request: UserRequest): # 获取或创建会话 session_id request.session_id or str(uuid.uuid4()) if session_id not in user_sessions: user_sessions[session_id] { agent: get_agent_for_user(session_id), profile: UserProfileManager(session_id) } session user_sessions[session_id] agent session[agent] profile_manager session[profile] # 1. 更新用户画像将本次对话作为记忆存储 profile_manager.add_memory( fUser said: {request.message}, memory_typeconversation ) # 2. 获取相关用户上下文 user_context profile_manager.get_user_context(request.message) # 3. 调用智能体 try: response agent.invoke({ input: request.message, user_context: user_context }) final_output response.get(output, I didnt get a response.) # 4. 将助手的回复也作为记忆存储可选 profile_manager.add_memory( fWorkBuddy replied: {final_output}, memory_typeconversation ) # 判断是否需要后续跟进例如任务未完成在等待用户输入参数 needs_follow_up please provide in final_output.lower() or ? in final_output return AgentResponse( replyfinal_output, session_idsession_id, needs_follow_upneeds_follow_up ) except Exception as e: raise HTTPException(status_code500, detailfAgent processing failed: {str(e)}) app.get(/session/{session_id}/memory) async def get_session_memory(session_id: str, query: Optional[str] None): 获取特定会话的记忆调试用 if session_id not in user_sessions: raise HTTPException(status_code404, detailSession not found) profile_manager user_sessions[session_id][profile] if query: memories profile_manager.query_memory(query) return {memories: [{content: doc.page_content, meta: doc.metadata} for doc, _ in memories]} else: # 返回最近的一些记忆简化处理 return {message: Provide a query parameter to search memories.}这个API提供了聊天接口和记忆查询接口。前端可以通过轮询或WebSocket与它交互。6.2 前端简单界面示例一个简单的Streamlit前端可以快速演示功能。# streamlit_app.py import streamlit as st import requests import json API_BASE_URL http://localhost:8000 # 假设后端运行在此 st.title( WorkBuddy - Your AI Assistant) st.caption(From Who I am to Get things done) # 初始化会话状态 if session_id not in st.session_state: st.session_state.session_id None if messages not in st.session_state: st.session_state.messages [] # 显示历史消息 for msg in st.session_state.messages: with st.chat_message(msg[role]): st.markdown(msg[content]) # 聊天输入 if prompt : st.chat_input(What can WorkBuddy do for you?): # 添加用户消息 st.session_state.messages.append({role: user, content: prompt}) with st.chat_message(user): st.markdown(prompt) # 调用后端API with st.chat_message(assistant): with st.spinner(WorkBuddy is thinking...): payload { message: prompt, session_id: st.session_state.session_id } try: response requests.post( f{API_BASE_URL}/chat, jsonpayload, timeout30 ) if response.status_code 200: data response.json() reply data[reply] st.session_state.session_id data[session_id] st.markdown(reply) st.session_state.messages.append({role: assistant, content: reply}) # 如果AI需要更多信息给出提示 if data.get(needs_follow_up): st.info(WorkBuddy is waiting for more information to complete the task.) else: st.error(fAPI Error: {response.text}) except requests.exceptions.RequestException as e: st.error(fFailed to connect to WorkBuddy backend: {e}) # 侧边栏会话管理 with st.sidebar: st.header(Session) if st.session_state.session_id: st.code(fSession ID: {st.session_state.session_id[:8]}...) if st.button(New Session): st.session_state.session_id None st.session_state.messages [] st.rerun() else: st.info(No active session. Start chatting to create one.)6.3 监控、评估与持续改进上线后必须建立监控和评估体系。日志记录记录每一次用户交互、AI的思考过程、工具调用参数和结果、最终输出。这用于调试和后续分析。关键指标任务完成率用户明确的任务中有多少被成功执行工具调用准确率AI选择的工具是否合适参数是否正确用户满意度通过简单的“赞/踩”按钮或后续调研收集。平均对话轮次完成一个任务需要多少轮对话轮次过多可能意味着规划能力或工具效率有问题。反馈循环将失败的案例如工具调用错误、用户不满意加入一个评估数据集。定期用这个数据集微调模型的提示词或者作为few-shot示例加入系统提示中。工具扩展根据用户最常请求但当前无法完成的任务优先级开发新的工具。例如如果很多用户问“把这份PPT转换成PDF”那就需要集成一个convert_ppt_to_pdf的工具。7. 常见问题与实战排坑指南在实际开发和测试中我遇到了不少典型问题这里总结一下。7.1 智能体陷入循环或执行无关操作这是ReAct模式最常见的问题。AI可能在一个步骤上不断重复或者调用一系列不直接解决问题的工具。解决方案严格限制迭代次数如之前代码中的max_iterations5强制跳出。改进提示词在系统提示中强调“如果当前步骤无法推进任务请停止并总结当前状况向用户求助”。后处理检查在最终输出前让另一个轻量级模型或规则检查本次对话是否实质性推进了用户目标。如果没有触发一个重置或澄清。示例学习在提示词中加入几个成功完成多步任务的示例让模型学会正确的规划模式。7.2 工具描述不准确导致误用如果工具的描述description过于宽泛或模糊AI会错误地调用它。踩坑案例早期我给文件搜索工具的描述是“搜索文件”结果用户问“明天的天气”AI也去调用文件搜索。优化方案将描述写得极其具体限定使用场景。例如“在用户指定的本地目录中基于文档内容语义搜索相关的文本和PDF文件。适用于当用户想查找包含某些特定主题或关键词的文档时。不适用于查询实时信息、天气、新闻或网络搜索。”7.3 处理模糊或开放式用户请求用户常说“帮我做一下那个事”或“整理一下资料”。AI需要主动澄清。实战技巧除了前面提到的独立规划器可以在主循环中设置一个“澄清状态”。当模型认为输入模糊时它输出一个特定的动作如Action: ask_for_clarification然后前端捕获到这个特殊动作弹出表单让用户填写缺失的字段如时间、文件类型、具体操作。这比让模型生成自然语言问题更结构化也更容易处理。7.4 上下文长度限制与记忆管理长对话会耗尽模型的上下文窗口导致忘记早期的关键信息。解决方案组合拳摘要记忆定期例如每10轮对话让模型对之前的对话历史进行总结将详细的对话压缩成几个要点存入长期记忆向量库然后从当前对话上下文中移除旧的历史。这称为“对话摘要”。重要性评分在存储记忆时让模型对记忆的重要性打分例如1-5分。在检索时优先召回高分记忆。用户明确说“记住这个”的信息可以手动打高分。分层记忆区分短期工作记忆最近几轮对话和长期档案记忆向量库。每次查询时同时从两者中获取信息。7.5 成本控制频繁调用大模型和嵌入模型成本可能快速上升。优化策略缓存对常见的、结果不变的查询如“公司的核心价值观是什么”的嵌入结果和AI回复进行缓存。小模型分工用便宜的小模型如GPT-3.5 Turbo处理简单的分类、路由或摘要任务只在复杂的规划和创意生成时用大模型如GPT-4。本地模型对于嵌入任务可以使用开源的本地模型如all-MiniLM-L6-v2避免调用OpenAI的Embedding API。对于简单的工具选择基于描述也可以训练一个小的分类模型。预算监控在代码中集成成本计算对每个API调用记录token消耗并设置每日/每月预算警报。构建WorkBuddy这样的智能体是一个持续迭代的过程。从最简单的“问答机器人”到能“干活”的伙伴中间需要不断地调试工具、优化提示词、完善安全策略。最深的体会是不要指望一个超级提示词解决所有问题。可靠的智能体背后是大量细致的工程工作清晰的工具定义、稳健的错误处理、合理的状态管理以及持续的用户反馈学习。当你看到它第一次正确理解一个复杂任务并自动调用一系列工具完成时那种成就感是巨大的。 从聊天到执行：基于LangChain构建智能体WorkBuddy的架构与实现
1. 项目概述一个“会干活”的AI伙伴最近在折腾AI应用落地的朋友估计都绕不开一个核心痛点如何让大语言模型从一个“博学的聊天对象”真正变成一个能帮你“干脏活累活”的可靠伙伴这就是“WorkBuddy”这个项目想解决的根本问题。它不是一个简单的聊天机器人而是一个旨在深度理解你的身份、习惯和任务上下文并主动替你执行具体操作的智能体。简单来说传统的AI助手你问它“今天天气怎么样”它会告诉你天气。但WorkBuddy的思路是当你早上说“今天有什么安排”时它不仅能基于日历告诉你上午十点有会还能根据你“总是提前15分钟到会议室”的习惯在9:45提醒你动身甚至自动帮你把会议资料从云盘下载到本地并打开相关的演示文稿。它的目标是从被动应答的“我是谁”身份识别与记忆跃迁到主动服务的“帮我干活”任务理解与自动化执行。这个项目适合所有对AI应用开发、智能体构建以及自动化工作流感兴趣的朋友无论是想提升个人效率的极客还是希望为企业内部打造智能助手的开发者。接下来我会拆解实现这样一个“WorkBuddy”的核心设计思路、关键技术栈并分享从零搭建原型过程中那些必须注意的“坑”。2. 核心架构设计让AI拥有“手”和“记忆”要实现从认知到执行的跨越系统的架构设计是关键。你不能只靠一个语言模型包打天下它需要一套组合拳。2.1 三层核心组件解析一个完整的WorkBuddy系统可以抽象为三层感知与理解层、决策与规划层、执行与反馈层。感知与理解层这是“我是谁”的基础。它的核心是用户画像与上下文管理。你需要一个向量数据库来存储和检索用户的历史对话、行为偏好、个人资料。例如用户提到过“我习惯用Markdown写周报”这句话需要被解析、嵌入成向量存起来。下次用户说“写周报”时系统不仅能检索到“周报”这个任务还能关联到“Markdown格式”这个偏好。这里常用的工具有ChromaDB、Pinecone或者Weaviate。关键在于设计好数据的嵌入和检索策略比如将用户信息、对话历史、任务结果分开存储通过元数据过滤提高检索精度。决策与规划层这是大脑。大语言模型在这里扮演核心角色但它不是直接输出答案而是输出一个可执行的计划。当用户说“帮我整理一下上周的项目资料”模型需要拆解任务1. 确定时间范围上周一至周日2. 定位资料存储位置可能是Google Drive的“ProjectX”文件夹3. 定义“整理”的具体操作按日期重命名文件、生成摘要目录4. 识别所需工具调用云盘API、调用文件处理API。这个过程被称为“任务分解”或“思维链规划”。我们通常使用像GPT-4、Claude 3或者开源的Llama 3、DeepSeek等具有较强推理能力的模型并通过精心设计的提示词来引导其进行结构化输出。执行与反馈层这是“手”。AI自己不能操作电脑它需要通过工具调用来实现。你需要为模型提供一套“工具集”比如read_file、search_web、send_email、execute_shell_command需极其谨慎等。模型根据规划层的输出决定调用哪个工具并生成正确的调用参数。执行后工具返回结果成功或失败附带数据这个结果会反馈给模型模型根据结果决定下一步是继续执行还是向用户请求澄清。这一层的实现框架ReAct和LangChain是很好的起点它们提供了将模型、工具、记忆连接起来的标准化模式。注意工具调用是安全风险最高的环节。绝对禁止开放高危工具如无条件执行任意Shell命令、直接访问数据库写操作给模型。必须实施严格的工具权限管控和参数校验。例如文件删除工具只能删除特定临时目录下的文件邮件发送工具需要二次确认或限制收件人域名。2.2 状态管理与会话持久化WorkBuddy需要记住跨会话的信息。这不仅仅是保存聊天记录那么简单而是维护一个动态的会话状态。这个状态包括当前正在执行的多步骤任务进度、已收集到的用户输入参数、临时生成的数据等。例如用户说“订一张明天去上海的机票”WorkBuddy进入订票流程状态变为awaiting_departure_time。你回复“下午出发”状态更新并接着问“您偏好哪个航空公司”。实现上可以将整个会话状态一个复杂的Python字典或Pydantic模型序列化后与用户ID关联存储到数据库如Redis用于快速会话缓存PostgreSQL用于长期持久化。每次交互开始前先加载状态这样模型就知道“对话进行到哪一步了”。LangChain的ConversationChain或自定义的StateGraph在LangGraph中是管理这类状态的强大工具。3. 关键技术实现与工具链选型理论讲完我们落到具体的代码和工具上。我会以构建一个简单的、能处理“文件整理”和“信息查询”任务的WorkBuddy原型为例。3.1 搭建基础环境与智能体核心首先我们选择LangChain作为智能体框架因为它生态丰富抽象得当。假设我们使用OpenAI的GPT-4作为核心模型。# 环境准备 pip install langchain langchain-openai langchain-community chromadb python-dotenv# core_agent.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain.memory import ConversationBufferMemory from langchain.agents import AgentExecutor, create_react_agent from langchain.tools import Tool from langchain.prompts import PromptTemplate from langchain_community.tools import DuckDuckGoSearchRun # 假设我们自定义了一些工具后面会定义 from custom_tools import FileSearchTool, SummarizeDocumentTool load_dotenv() # 1. 初始化大模型 llm ChatOpenAI( modelgpt-4-turbo-preview, temperature0.1, # 低温度保证任务执行的稳定性 api_keyos.getenv(OPENAI_API_KEY) ) # 2. 初始化记忆用于存储对话历史 memory ConversationBufferMemory( memory_keychat_history, return_messagesTrue, output_keyoutput ) # 3. 定义工具集 search DuckDuckGoSearchRun() file_search FileSearchTool() summarizer SummarizeDocumentTool() tools [ Tool( nameWeb Search, funcsearch.run, descriptionUseful for when you need to answer questions about current events or general knowledge. Input should be a clear search query. ), Tool( nameFile Search, funcfile_search.run, descriptionUseful for searching and retrieving information from the users document repository. Input should be keywords related to the file content. ), Tool( nameDocument Summarizer, funcsummarizer.run, descriptionUseful for summarizing long documents. Input should be the path to a text file or the text content itself. ) ] # 4. 创建ReAct智能体 prompt PromptTemplate.from_template( You are WorkBuddy, a helpful assistant that can perform tasks. You have access to the following tools: {tools} Use the following format: Question: the input question you must answer Thought: you should always think about what to do Action: the action to take, should be one of [{tool_names}] Action Input: the input to the action Observation: the result of the action ... (this Thought/Action/Action Input/Observation can repeat N times) Thought: I now know the final answer Final Answer: the final answer to the original input Begin! Previous conversation history: {chat_history} Question: {input} Thought:{agent_scratchpad} ) agent create_react_agent(llm, tools, prompt) agent_executor AgentExecutor( agentagent, toolstools, memorymemory, verboseTrue, # 开发时打开查看思考过程 handle_parsing_errorsTrue, max_iterations5 # 防止死循环 ) # 运行示例 if __name__ __main__: response agent_executor.invoke({input: 帮我找一下上个月关于市场分析的文档并总结核心观点。}) print(response[output])这个代码搭建了一个最基本的、具备思考Thought和行动Action能力的智能体。它可以通过搜索工具找文件然后用总结工具生成摘要。3.2 实现自定义工具以文件操作为例上面用到的FileSearchTool和SummarizeDocumentTool需要我们自己实现。这里展示一个连接简单本地文件系统的搜索工具。# custom_tools.py import os from pathlib import Path from typing import Type from langchain.tools import BaseTool from pydantic import BaseModel, Field from langchain_community.document_loaders import TextLoader, PyPDFLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings class FileSearchInput(BaseModel): query: str Field(descriptionKeywords to search for within documents) directory: str Field(default., descriptionDirectory to search in) class FileSearchTool(BaseTool): name file_search description Searches for content within text and PDF files in a specified directory. args_schema: Type[BaseModel] FileSearchInput return_direct: bool False # 结果返回给Agent继续处理 vector_store None def _setup_vector_store(self, directory: str): 初始化或加载指定目录的文档向量库 persist_path f./chroma_db/{os.path.basename(directory)} embeddings OpenAIEmbeddings() if os.path.exists(persist_path): # 加载已有库 self.vector_store Chroma( persist_directorypersist_path, embedding_functionembeddings ) else: # 构建新库 documents [] for ext in [*.txt, *.md, *.pdf]: for file_path in Path(directory).rglob(ext): try: if file_path.suffix .pdf: loader PyPDFLoader(str(file_path)) else: loader TextLoader(str(file_path), encodingutf-8) loaded_docs loader.load() # 为每个文档添加来源元数据 for doc in loaded_docs: doc.metadata[source] str(file_path) doc.metadata[filename] file_path.name documents.extend(loaded_docs) except Exception as e: print(fError loading {file_path}: {e}) if documents: # 分割文本便于检索 text_splitter RecursiveCharacterTextSplitter( chunk_size1000, chunk_overlap200 ) split_docs text_splitter.split_documents(documents) # 创建向量存储 self.vector_store Chroma.from_documents( documentssplit_docs, embeddingembeddings, persist_directorypersist_path ) self.vector_store.persist() else: self.vector_store None def _run(self, query: str, directory: str .) - str: if not query: return Please provide a search query. # 懒加载向量库 if self.vector_store is None: self._setup_vector_store(directory) if self.vector_store is None: return fNo indexable text or PDF files found in {directory}. # 执行相似性搜索 try: results self.vector_store.similarity_search_with_relevance_scores(query, k3) if not results: return No relevant documents found for your query. response Here are the most relevant documents I found:\n for i, (doc, score) in enumerate(results): response f\n{i1}. **{doc.metadata.get(filename, Unknown)}** (Relevance: {score:.2f})\n response f Source: {doc.metadata.get(source)}\n response f Snippet: {doc.page_content[:200]}...\n return response except Exception as e: return fAn error occurred during search: {str(e)} async def _arun(self, query: str, directory: str .): raise NotImplementedError(Async operation not supported yet.)这个工具做了几件关键事1. 它接受查询和目录参数2. 首次使用时会加载目录下的文档并构建向量索引这是一个耗时的初始化过程实际产品中需要异步或后台处理3. 执行语义搜索返回最相关的文档片段及其来源。这比简单文件名搜索强大得多因为它理解内容。3.3 用户画像与长期记忆的实现让WorkBuddy记住“我是谁”我们需要一个用户画像系统。一个简单的实现可以基于向量数据库存储用户相关的所有信息片段。# user_profile.py from langchain_community.vectorstores import Chroma from langchain_openai import OpenAIEmbeddings from langchain.schema import Document from datetime import datetime import json class UserProfileManager: def __init__(self, user_id: str): self.user_id user_id self.profile_store_path f./user_profiles/{user_id} self.embeddings OpenAIEmbeddings() self._init_store() def _init_store(self): 初始化或加载用户的向量化记忆库 try: self.vector_store Chroma( persist_directoryself.profile_store_path, embedding_functionself.embeddings ) except: # 首次创建 self.vector_store Chroma.from_documents( documents[], embeddingself.embeddings, persist_directoryself.profile_store_path ) def add_memory(self, memory_text: str, memory_type: str conversation, metadata: dict None): 添加一段记忆到用户画像 if metadata is None: metadata {} metadata.update({ user_id: self.user_id, type: memory_type, timestamp: datetime.now().isoformat() }) doc Document(page_contentmemory_text, metadatametadata) self.vector_store.add_documents([doc]) self.vector_store.persist() def query_memory(self, query: str, memory_type: str None, k: int 3): 查询与当前问题相关的用户记忆 filter_dict {user_id: self.user_id} if memory_type: filter_dict[type] memory_type results self.vector_store.similarity_search_with_relevance_scores( query, kk, filterfilter_dict ) return results def get_user_context(self, current_query: str) - str: 获取与当前查询相关的用户上下文用于拼接到提示词中 relevant_memories self.query_memory(current_query) if not relevant_memories: return No specific user context found. context_str Relevant user context from past interactions:\n for i, (doc, score) in enumerate(relevant_memories): context_str f- {doc.page_content} (from {doc.metadata.get(type, unknown)})\n return context_str # 使用示例 if __name__ __main__: user_manager UserProfileManager(user_123) # 添加一些记忆 user_manager.add_memory( The user prefers to receive summaries in bullet points., memory_typepreference ) user_manager.add_memory( The user is working on a project called Phoenix related to market analysis., memory_typeproject ) # 查询 context user_manager.get_user_context(How should I format the report?) print(context)这样每次用户交互时我们可以先调用get_user_context获取相关历史信息并将其作为系统提示词的一部分让模型在了解用户背景的情况下进行回复和规划。4. 提示词工程引导AI正确规划与执行智能体的“智商”很大程度上取决于提示词。对于WorkBuddy我们需要设计多阶段的提示词。4.1 系统提示词设计系统提示词定义了AI的角色、能力和行为规范。这是最核心的部分。你是一个名为WorkBuddy的智能助手。你的核心目标是理解用户需求并主动、准确地完成任务。 **你的身份与原则** 1. 你是用户的专属助手熟知用户的偏好和历史上下文会提供给你。 2. 你拥有工具使用能力。在回答问题或执行任务时应优先考虑是否可以通过调用工具更高效、准确地完成。 3. 对于复杂任务你必须先制定分步计划然后逐步执行。每次只执行一步观察结果后再决定下一步。 4. 如果工具执行失败或结果不明确不要猜测。向用户澄清或尝试替代方案。 5. 始终以帮助用户完成具体工作为目标回答应简洁、务实避免冗长的理论阐述。 **可用的工具** {tools} **当前用户上下文** {user_context} **对话历史** {chat_history} 请严格按照以下格式输出 Thought: 分析用户请求结合上下文和历史。决定是否需要使用工具以及使用哪个工具。 Action: 需要使用的工具名如果不使用工具则为 None。 Action Input: 工具的输入参数如果不使用工具则为 None。 Observation: 工具返回的结果第一步先留空。 ...重复 Thought/Action/Action Input/Observation 直到任务完成或无需更多工具 Thought: 我现在有足够信息给出最终答案。 Final Answer: 对用户的最终回复汇总结果或确认任务完成。这个提示词明确了ReAct格式并强调了规划、工具优先和基于上下文的行动。4.2 任务分解与参数提取提示词对于模糊的用户指令我们需要一个专门的步骤来澄清和分解。这可以通过一个独立的“规划器”LLM调用来实现。# planner_prompt.py from langchain.prompts import ChatPromptTemplate from langchain_core.output_parsers import JsonOutputParser from pydantic import BaseModel, Field from typing import List class Subtask(BaseModel): description: str Field(description对该子任务的清晰描述) tool: str Field(description完成此任务所需的工具名称如 FileSearchTool, Web Search, 或 None) parameters: dict Field(default_factorydict, description调用工具所需的参数) class TaskPlan(BaseModel): main_goal: str Field(description用户的主要目标) subtasks: List[Subtask] Field(description为达成目标所需的子任务序列) needs_clarification: bool Field(description是否需要向用户询问更多信息) clarification_question: str Field(default, description如果需要澄清要问什么问题) planner_prompt ChatPromptTemplate.from_messages([ (system, 你是一个任务规划专家。请将用户的模糊请求分解为具体、可执行的子任务步骤。 考虑可用的工具{tool_descriptions}。 如果用户请求缺少必要信息如时间、地点、具体文件名请标记需要澄清并生成一个清晰的问题。 输出必须是有效的JSON格式。), (human, 用户请求{user_input}\n当前已知用户上下文{user_context}) ]) def create_plan(user_input: str, user_context: str, llm): parser JsonOutputParser(pydantic_objectTaskPlan) full_prompt planner_prompt.format_parts( user_inputuser_input, user_contextuser_context, tool_descriptions, .join([f{t.name}: {t.description} for t in tools]) # 假设tools已定义 ) message full_prompt.to_messages() response llm.invoke(message) plan_dict parser.parse(response.content) return TaskPlan(**plan_dict)这个规划器会先于主智能体运行。如果needs_clarification为TrueWorkBuddy会先问用户那个问题。否则它就获得了一个清晰的子任务列表可以按顺序执行或交由智能体动态调整。5. 安全、隐私与工程化考量将AI智能体投入实际使用尤其是处理用户文件和数据时安全和隐私是生命线。5.1 工具执行沙箱与权限控制绝对不能允许AI直接、无限制地调用系统命令或访问所有文件。必须实现一个安全层。# security_layer.py import subprocess import os from pathlib import Path class SafeCommandTool: 一个安全的、受限的Shell命令执行工具 allowed_commands { list_dir: {cmd: [ls, -la], cwd: None}, current_date: {cmd: [date], cwd: None}, # 可以定义更多白名单命令 } allowed_directories [/tmp/workbuddy, os.path.expanduser(~/workbuddy_safe)] staticmethod def run(command_name: str, args: list None, working_dir: str None): if command_name not in SafeCommandTool.allowed_commands: return fError: Command {command_name} is not in the allowed list. config SafeCommandTool.allowed_commands[command_name].copy() cmd_list config[cmd] if args: cmd_list.extend(args) # 限制工作目录 safe_cwd working_dir if working_dir in SafeCommandTool.allowed_directories else config.get(cwd) if safe_cwd: Path(safe_cwd).mkdir(parentsTrue, exist_okTrue) try: result subprocess.run( cmd_list, cwdsafe_cwd, capture_outputTrue, textTrue, timeout10 ) if result.returncode 0: return result.stdout else: return fCommand failed with error: {result.stderr} except subprocess.TimeoutExpired: return Error: Command timed out. except Exception as e: return fError executing command: {str(e)}这个工具只允许执行预定义的白名单命令并在指定目录下运行。对于文件操作同样需要路径白名单或沙箱机制。5.2 数据隐私与匿名化处理如果WorkBuddy需要处理包含个人身份信息的数据必须在发送给外部AI API如OpenAI前进行匿名化。# privacy_filter.py import re class PrivacyFilter: patterns { email: r\b[A-Za-z0-9._%-][A-Za-z0-9.-]\.[A-Z|a-z]{2,}\b, phone_cn: r\b1[3-9]\d{9}\b, # 简单中国手机号匹配 id_card: r\b[1-9]\d{5}(18|19|20)\d{2}(0[1-9]|1[0-2])(0[1-9]|[12]\d|3[01])\d{3}[0-9Xx]\b # 简单身份证号匹配 } staticmethod def anonymize_text(text: str): anonymized text for ptype, pattern in PrivacyFilter.patterns.items(): anonymized re.sub(pattern, f[{ptype}_redacted], anonymized) return anonymized staticmethod def should_send_to_external_api(text: str, user_consent: bool): 决定文本是否可发送外部API if not user_consent: return False, No user consent for external API. # 检查是否包含高敏感信息可自定义规则 high_sensitivity_pattern r(password|token|key|secret)\s*[:]\s*\S if re.search(high_sensitivity_pattern, text, re.IGNORECASE): return False, Text contains high-sensitivity information. # 匿名化后发送 safe_text PrivacyFilter.anonymize_text(text) return True, safe_text在调用LLM API前先通过should_send_to_external_api检查并处理文本。同时所有本地存储的用户数据如向量库应进行加密。5.3 错误处理与鲁棒性增强智能体在复杂环境中必然会出错。健壮的错误处理机制必不可少。# error_handling.py from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import openai class RobustAgentExecutor: def __init__(self, agent_executor): self.agent agent_executor retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10), retryretry_if_exception_type((openai.APITimeoutError, openai.APIConnectionError)) ) def invoke_with_retry(self, input_dict): 带重试的调用主要处理网络或API瞬时错误 try: return self.agent.invoke(input_dict) except Exception as e: # 解析智能体特定的错误如工具调用失败、输出解析错误 error_msg str(e) if Could not parse LLM output in error_msg: # 模型输出不符合格式可能是复杂问题导致混乱 return { output: I encountered an issue while planning the steps. Could you please rephrase your request or break it down into simpler tasks? } elif Tool not found in error_msg: return { output: I tried to use a tool that isnt available. Let me try a different approach. } else: # 其他未预见的错误记录并返回友好信息 print(fUnhandled agent error: {error_msg}) return { output: An unexpected error occurred. Please try again or contact support if the problem persists. }此外要为每个工具调用设置超时并为智能体的“思考循环”设置最大迭代次数防止陷入死循环消耗资源。6. 部署与迭代从原型到可用产品让WorkBuddy真正可用需要把它包装成一个服务。6.1 后端API服务搭建使用FastAPI可以快速构建一个提供Web服务的后端。# main.py (FastAPI 后端) from fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel from typing import Optional import uuid from agent_core import get_agent_for_user # 假设这是封装好的智能体获取函数 from user_profile import UserProfileManager app FastAPI(titleWorkBuddy API) # 简单的内存会话存储生产环境用Redis user_sessions {} class UserRequest(BaseModel): message: str session_id: Optional[str] None class AgentResponse(BaseModel): reply: str session_id: str needs_follow_up: bool False app.post(/chat, response_modelAgentResponse) async def chat_with_buddy(request: UserRequest): # 获取或创建会话 session_id request.session_id or str(uuid.uuid4()) if session_id not in user_sessions: user_sessions[session_id] { agent: get_agent_for_user(session_id), profile: UserProfileManager(session_id) } session user_sessions[session_id] agent session[agent] profile_manager session[profile] # 1. 更新用户画像将本次对话作为记忆存储 profile_manager.add_memory( fUser said: {request.message}, memory_typeconversation ) # 2. 获取相关用户上下文 user_context profile_manager.get_user_context(request.message) # 3. 调用智能体 try: response agent.invoke({ input: request.message, user_context: user_context }) final_output response.get(output, I didnt get a response.) # 4. 将助手的回复也作为记忆存储可选 profile_manager.add_memory( fWorkBuddy replied: {final_output}, memory_typeconversation ) # 判断是否需要后续跟进例如任务未完成在等待用户输入参数 needs_follow_up please provide in final_output.lower() or ? in final_output return AgentResponse( replyfinal_output, session_idsession_id, needs_follow_upneeds_follow_up ) except Exception as e: raise HTTPException(status_code500, detailfAgent processing failed: {str(e)}) app.get(/session/{session_id}/memory) async def get_session_memory(session_id: str, query: Optional[str] None): 获取特定会话的记忆调试用 if session_id not in user_sessions: raise HTTPException(status_code404, detailSession not found) profile_manager user_sessions[session_id][profile] if query: memories profile_manager.query_memory(query) return {memories: [{content: doc.page_content, meta: doc.metadata} for doc, _ in memories]} else: # 返回最近的一些记忆简化处理 return {message: Provide a query parameter to search memories.}这个API提供了聊天接口和记忆查询接口。前端可以通过轮询或WebSocket与它交互。6.2 前端简单界面示例一个简单的Streamlit前端可以快速演示功能。# streamlit_app.py import streamlit as st import requests import json API_BASE_URL http://localhost:8000 # 假设后端运行在此 st.title( WorkBuddy - Your AI Assistant) st.caption(From Who I am to Get things done) # 初始化会话状态 if session_id not in st.session_state: st.session_state.session_id None if messages not in st.session_state: st.session_state.messages [] # 显示历史消息 for msg in st.session_state.messages: with st.chat_message(msg[role]): st.markdown(msg[content]) # 聊天输入 if prompt : st.chat_input(What can WorkBuddy do for you?): # 添加用户消息 st.session_state.messages.append({role: user, content: prompt}) with st.chat_message(user): st.markdown(prompt) # 调用后端API with st.chat_message(assistant): with st.spinner(WorkBuddy is thinking...): payload { message: prompt, session_id: st.session_state.session_id } try: response requests.post( f{API_BASE_URL}/chat, jsonpayload, timeout30 ) if response.status_code 200: data response.json() reply data[reply] st.session_state.session_id data[session_id] st.markdown(reply) st.session_state.messages.append({role: assistant, content: reply}) # 如果AI需要更多信息给出提示 if data.get(needs_follow_up): st.info(WorkBuddy is waiting for more information to complete the task.) else: st.error(fAPI Error: {response.text}) except requests.exceptions.RequestException as e: st.error(fFailed to connect to WorkBuddy backend: {e}) # 侧边栏会话管理 with st.sidebar: st.header(Session) if st.session_state.session_id: st.code(fSession ID: {st.session_state.session_id[:8]}...) if st.button(New Session): st.session_state.session_id None st.session_state.messages [] st.rerun() else: st.info(No active session. Start chatting to create one.)6.3 监控、评估与持续改进上线后必须建立监控和评估体系。日志记录记录每一次用户交互、AI的思考过程、工具调用参数和结果、最终输出。这用于调试和后续分析。关键指标任务完成率用户明确的任务中有多少被成功执行工具调用准确率AI选择的工具是否合适参数是否正确用户满意度通过简单的“赞/踩”按钮或后续调研收集。平均对话轮次完成一个任务需要多少轮对话轮次过多可能意味着规划能力或工具效率有问题。反馈循环将失败的案例如工具调用错误、用户不满意加入一个评估数据集。定期用这个数据集微调模型的提示词或者作为few-shot示例加入系统提示中。工具扩展根据用户最常请求但当前无法完成的任务优先级开发新的工具。例如如果很多用户问“把这份PPT转换成PDF”那就需要集成一个convert_ppt_to_pdf的工具。7. 常见问题与实战排坑指南在实际开发和测试中我遇到了不少典型问题这里总结一下。7.1 智能体陷入循环或执行无关操作这是ReAct模式最常见的问题。AI可能在一个步骤上不断重复或者调用一系列不直接解决问题的工具。解决方案严格限制迭代次数如之前代码中的max_iterations5强制跳出。改进提示词在系统提示中强调“如果当前步骤无法推进任务请停止并总结当前状况向用户求助”。后处理检查在最终输出前让另一个轻量级模型或规则检查本次对话是否实质性推进了用户目标。如果没有触发一个重置或澄清。示例学习在提示词中加入几个成功完成多步任务的示例让模型学会正确的规划模式。7.2 工具描述不准确导致误用如果工具的描述description过于宽泛或模糊AI会错误地调用它。踩坑案例早期我给文件搜索工具的描述是“搜索文件”结果用户问“明天的天气”AI也去调用文件搜索。优化方案将描述写得极其具体限定使用场景。例如“在用户指定的本地目录中基于文档内容语义搜索相关的文本和PDF文件。适用于当用户想查找包含某些特定主题或关键词的文档时。不适用于查询实时信息、天气、新闻或网络搜索。”7.3 处理模糊或开放式用户请求用户常说“帮我做一下那个事”或“整理一下资料”。AI需要主动澄清。实战技巧除了前面提到的独立规划器可以在主循环中设置一个“澄清状态”。当模型认为输入模糊时它输出一个特定的动作如Action: ask_for_clarification然后前端捕获到这个特殊动作弹出表单让用户填写缺失的字段如时间、文件类型、具体操作。这比让模型生成自然语言问题更结构化也更容易处理。7.4 上下文长度限制与记忆管理长对话会耗尽模型的上下文窗口导致忘记早期的关键信息。解决方案组合拳摘要记忆定期例如每10轮对话让模型对之前的对话历史进行总结将详细的对话压缩成几个要点存入长期记忆向量库然后从当前对话上下文中移除旧的历史。这称为“对话摘要”。重要性评分在存储记忆时让模型对记忆的重要性打分例如1-5分。在检索时优先召回高分记忆。用户明确说“记住这个”的信息可以手动打高分。分层记忆区分短期工作记忆最近几轮对话和长期档案记忆向量库。每次查询时同时从两者中获取信息。7.5 成本控制频繁调用大模型和嵌入模型成本可能快速上升。优化策略缓存对常见的、结果不变的查询如“公司的核心价值观是什么”的嵌入结果和AI回复进行缓存。小模型分工用便宜的小模型如GPT-3.5 Turbo处理简单的分类、路由或摘要任务只在复杂的规划和创意生成时用大模型如GPT-4。本地模型对于嵌入任务可以使用开源的本地模型如all-MiniLM-L6-v2避免调用OpenAI的Embedding API。对于简单的工具选择基于描述也可以训练一个小的分类模型。预算监控在代码中集成成本计算对每个API调用记录token消耗并设置每日/每月预算警报。构建WorkBuddy这样的智能体是一个持续迭代的过程。从最简单的“问答机器人”到能“干活”的伙伴中间需要不断地调试工具、优化提示词、完善安全策略。最深的体会是不要指望一个超级提示词解决所有问题。可靠的智能体背后是大量细致的工程工作清晰的工具定义、稳健的错误处理、合理的状态管理以及持续的用户反馈学习。当你看到它第一次正确理解一个复杂任务并自动调用一系列工具完成时那种成就感是巨大的。