在实际 AI 应用开发中构建一个能理解指令、调用工具并完成复杂任务的智能体Agent已成为从原型走向实用产品的关键一步。无论是基于大语言模型LLM的自动化助手还是结合数字人形象的交互前端或是利用 LangChain 框架搭建的问答系统其核心都在于如何让 AI 模型具备规划、决策和执行的能力。对于希望进入 AI 应用开发领域的工程师而言掌握 Agent 的开发范式意味着能够将 LLM 的通用能力与具体业务逻辑、外部工具和用户界面深度结合创造出真正有价值的解决方案。本文将从零开始带你构建一个覆盖 LLM 集成、数字人交互、LangChain 框架应用、问答系统搭建以及商业级智能体设计的实战项目。我们将聚焦于工程实现解释每一步背后的设计逻辑并提供可复现的代码、配置和排错指南。无论你是希望将 LLM 能力接入现有系统还是想打造一个具备自主行动能力的 AI 助手这篇文章都将为你提供一条清晰的实践路径。1. 理解 Agent 的核心从 LLM 到可执行动作在深入代码之前必须厘清几个核心概念。Agent 并非一个神秘的黑盒其本质是一个决策循环系统它接收用户输入利用 LLM 进行思考规划选择并调用合适的工具行动最后将工具执行结果整合后返回给用户观察。这个“思考-行动-观察”的循环是 Agent 区别于简单聊天机器人的关键。1.1 LLM 作为 Agent 的“大脑”LLM 是 Agent 的推理核心。它不直接操作世界而是通过生成文本如 JSON 格式的指令来指导外部工具执行动作。因此为 LLM 提供清晰、结构化的提示Prompt至关重要。一个典型的 Agent Prompt 需要包含系统角色定义告诉模型它现在是一个具备特定能力的助手。可用工具描述清晰列出每个工具的名称、功能、输入参数格式。输出格式要求强制模型以指定格式如 JSON输出便于程序解析。思考过程示例通过少量示例Few-shot引导模型进行链式推理。1.2 工具Tools作为 Agent 的“手脚”工具是 Agent 与外部世界交互的接口。它可以是一个计算器函数、一个数据库查询 API、一个网页搜索接口甚至是控制数字人表情的指令。工具的设计需要满足两个条件一是功能明确输入输出定义清晰二是能被 LLM 理解即其描述必须转化为自然语言并嵌入到提示词中。1.3 框架如 LangChain作为 Agent 的“骨架”手动管理提示词、解析模型输出、维护工具调用状态是非常繁琐且容易出错的。因此我们使用 LangChain 这类框架。它抽象了 Agent 的核心组件提供了标准化的方式来定义工具Tool。构建提示模板PromptTemplate。创建 Agent 执行器AgentExecutor。管理对话历史Memory。理解了这些我们就知道构建 Agent 的实质是选择合适的 LLM用框架定义好工具和流程然后编写代码将各部分连接起来。2. 环境准备与核心依赖配置在开始任何项目之前确保开发环境一致是避免后续诡异问题的前提。以下配置基于 Python 环境这是当前 AI 应用开发最主流的生态。2.1 Python 与包管理工具建议使用 Python 3.10 或 3.11这两个版本在兼容性和稳定性上最为平衡。使用conda或venv创建独立的虚拟环境。# 创建并激活 conda 环境推荐 conda create -n ai-agent python3.10 conda activate ai-agent # 或者使用 venv python -m venv venv # Windows venv\Scripts\activate # Linux/Mac source venv/bin/activate2.2 安装核心依赖库我们将使用pip安装项目所需的核心库。请根据你的网络情况选择合适的镜像源。# 升级 pip pip install --upgrade pip # 安装 LangChain 及其社区工具包、OpenAI SDK用于调用 GPT 等模型 pip install langchain langchain-community openai # 安装用于本地 LLM 集成的库例如使用 Ollama 或 llama.cpp pip install ollama # 安装 Web 框架用于构建 API 服务 pip install fastapi uvicorn # 安装向量数据库客户端用于构建 RAG 问答系统 pip install chromadb # 安装数字人相关 SDK以某云服务为例此处为示例需替换为实际 SDK # pip install some-digital-human-sdk注意数字人 SDK 高度依赖于所选服务商如腾讯云、阿里云、硅基等安装命令和 API 调用方式各不相同。本文后续将以一个模拟的客户端类为例讲解集成逻辑实际开发请查阅对应服务商的官方文档。2.3 配置 API 密钥与环境变量许多服务如 OpenAI、向量数据库、数字人平台都需要 API 密钥。永远不要将密钥硬编码在代码中。创建.env文件# .env OPENAI_API_KEYsk-your-openai-key-here # 如果使用其他模型服务如通义千问、DeepSeek 等 DASHSCOPE_API_KEYyour-dashscope-key # 数字人服务密钥 DIGITAL_HUMAN_APP_IDyour_app_id DIGITAL_HUMAN_APP_KEYyour_app_key在 Python 代码中加载环境变量推荐使用python-dotenv库。pip install python-dotenv# config.py import os from dotenv import load_dotenv load_dotenv() # 加载 .env 文件中的变量到环境变量 OPENAI_API_KEY os.getenv(OPENAI_API_KEY) if not OPENAI_API_KEY: raise ValueError(请在 .env 文件中设置 OPENAI_API_KEY)3. 项目一构建你的第一个 LangChain Agent让我们从一个最简单的 Agent 开始一个能进行数学计算和当前日期查询的助手。这个项目将贯穿 LangChain 的核心概念。3.1 定义工具Tools工具是 Agent 能力的扩展。我们先定义两个简单的工具。# tools/calculator_tool.py from langchain.tools import tool from datetime import datetime tool def calculate(expression: str) - str: 用于执行数学计算。输入是一个数学表达式字符串例如 ‘(12 5) * 2‘。 try: # 警告在生产环境中直接使用 eval 是极其危险的这里仅用于演示。 # 应使用安全的表达式求值库如 asteval。 result eval(expression, {__builtins__: {}}, {}) return f计算结果: {result} except Exception as e: return f计算错误: {e} tool def get_current_time(query: str) - str: 当用户询问当前时间、日期或今天星期几时使用此工具。输入是用户关于时间的问题。 now datetime.now() # 根据问题返回不同格式的时间 if 日期 in query or 几号 in query: return f当前日期是: {now.strftime(%Y年%m月%d日)} elif 星期 in query: return f今天是: {now.strftime(%A)} else: return f当前时间是: {now.strftime(%H:%M:%S)}3.2 创建 Agent 执行器我们将使用 OpenAI 的 GPT-3.5-turbo 作为 LLM并组合上面定义的工具。# agent/basic_agent.py import os from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder from tools.calculator_tool import calculate, get_current_time # 1. 初始化 LLM llm ChatOpenAI( modelgpt-3.5-turbo-1106, # 或 gpt-4 temperature0, # 降低随机性使 Agent 行为更确定 openai_api_keyos.getenv(OPENAI_API_KEY) ) # 2. 准备工具列表 tools [calculate, get_current_time] # 3. 构建提示模板 prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的助手可以回答问题和使用工具。 如果你需要计算或查询当前时间请使用提供的工具。 请严格按照工具要求的格式输入。 当你使用工具并获得结果后请用自然语言将结果整合到你的回答中。), MessagesPlaceholder(variable_namechat_history), # 预留位置给对话历史 (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), # 预留位置给 Agent 的思考过程 ]) # 4. 创建 Agent agent create_openai_tools_agent(llmllm, toolstools, promptprompt) # 5. 创建执行器并开启详细日志以便调试 agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, # 设置为 True 会在控制台打印详细的思考过程 handle_parsing_errorsTrue, # 当模型输出无法解析时尝试让模型重试 ) # 6. 运行 Agent if __name__ __main__: while True: user_input input(\n用户: ) if user_input.lower() in [退出, exit, quit]: break try: response agent_executor.invoke({input: user_input, chat_history: []}) print(f助手: {response[output]}) except Exception as e: print(f执行出错: {e})运行这个脚本尝试提问“123 乘以 456 等于多少” 或 “今天星期几”。在控制台你将看到verboseTrue带来的详细输出包括模型决定调用哪个工具、工具输入是什么、工具返回结果是什么以及模型如何组织最终回答。这是理解 Agent 工作流最直观的方式。3.3 关键配置解析与常见坑temperature参数对于工具调用类 Agent通常设置为 0 或接近 0如 0.1以确保模型输出稳定、可解析的指令。过高的温度会导致模型输出随机性大工具调用失败率高。工具描述tool装饰器下的函数文档字符串docstring至关重要。LLM 完全依赖这段描述来理解工具的功能和输入格式。描述必须清晰、准确。handle_parsing_errors当模型没有输出预期的 JSON 或格式错误时此设置允许执行器尝试让模型“重试”。在生产环境中你需要更精细的错误处理例如记录日志并返回友好提示。安全性警告示例中的calculate工具使用了eval这在实际项目中是绝对禁止的因为它会执行任意代码。此处仅用于演示工具定义的概念。真实场景应使用ast.literal_eval或专门的数学表达式解析库。常见问题排查表问题现象可能原因检查与解决报错openai.AuthenticationErrorAPI 密钥未设置或错误。1. 检查.env文件是否存在且格式正确。2. 检查环境变量是否成功加载 (print(os.getenv(“OPENAI_API_KEY”)))。3. 确认密钥是否有余额或权限。Agent 不调用工具直接回答问题。1. 工具描述不够清晰。2. 系统提示词未强调使用工具。3. 问题太简单模型觉得无需工具。1. 优化工具描述明确使用场景。2. 在系统提示词中强制要求“必须使用工具”。3. 使用verboseTrue查看模型思考过程。报错OutputParserException模型输出无法被解析为工具调用。1. 设置handle_parsing_errorsTrue。2. 检查提示词中是否明确要求了输出格式LangChain 默认已处理。3. 尝试换用更新或能力更强的模型如 gpt-4。4. 项目二集成数字人前端打造可视化交互单纯的命令行交互体验有限。我们将为 Agent 添加一个数字人前端使其能够通过语音和形象与用户交互。这里我们以模拟一个数字人驱动服务为例。4.1 设计交互流程目标用户通过 Web 页面与数字人对话数字人接收文本后由背后的 Agent 处理并生成回复文本再将文本驱动数字人播报出来。用户语音/文本 - Web前端 - [FastAPI后端] - LangChain Agent - 生成回复文本 - 数字人TTS服务 - 生成音频/动画 - 返回前端播放4.2 构建 FastAPI 后端服务# app/main.py from fastapi import FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel import asyncio from agent.basic_agent import agent_executor # 导入之前创建的 Agent from services.digital_human import DigitalHumanService # 模拟的数字人服务 app FastAPI(titleAI Agent with Digital Human) # 允许前端跨域请求 app.add_middleware( CORSMiddleware, allow_origins[*], # 生产环境应指定具体域名 allow_credentialsTrue, allow_methods[*], allow_headers[*], ) # 模拟的数字人服务类 class DigitalHumanService: def __init__(self): # 这里应初始化真实 SDK 的 Client # self.client SomeRealSDK(app_id, app_key) pass async def text_to_speech(self, text: str) - dict: 调用数字人服务将文本转为语音和驱动数据。 返回示例{“audio_url”: “http://...”, “animation_data”: {...}} # 模拟处理延迟 await asyncio.sleep(0.5) # 真实情况下这里调用服务商 API # response self.client.synthesize(texttext, ...) return { “audio_url”: f“https://example.com/audio/{hash(text)}.mp3”, # 模拟音频 URL “animation_data”: {“viseme”: “happy”, “duration”: len(text) * 0.05} # 模拟口型数据 } dh_service DigitalHumanService() class UserQuery(BaseModel): text: str session_id: str “default” # 用于维持对话会话 app.post(“/chat”) async def chat_with_agent(query: UserQuery): 接收用户文本通过 Agent 处理并驱动数字人响应。 try: # 1. Agent 处理 agent_response agent_executor.invoke({ “input”: query.text, “chat_history”: [] # 简单示例未实现历史管理 }) reply_text agent_response[“output”] # 2. 驱动数字人生成语音 dh_result await dh_service.text_to_speech(reply_text) return { “success”: True, “reply_text”: reply_text, “digital_human”: dh_result } except Exception as e: raise HTTPException(status_code500, detailf“处理请求时出错: {str(e)}”) if __name__ “__main__”: import uvicorn uvicorn.run(app, host“0.0.0.0”, port8000)4.3 简单前端页面HTML JavaScript创建一个templates/index.html文件用于输入文本和播放数字人响应。!DOCTYPE html html head title数字人 Agent 演示/title /head body h1与数字人助手对话/h1 div input type“text” id“userInput” placeholder“请输入您的问题…” style“width: 300px;” button onclick“sendMessage()”发送/button /div div h3助手回复/h3 p id“replyText”/p audio id“audioPlayer” controls style“display:none;”/audio div id“animationPlaceholder” style“margin-top:20px;” !-- 此处可嵌入数字人形象例如使用 Canvas 或 iframe -- p[数字人形象区域]/p /div /div script async function sendMessage() { const inputElem document.getElementById(‘userInput’); const text inputElem.value.trim(); if (!text) return; const replyElem document.getElementById(‘replyText’); const audioElem document.getElementById(‘audioPlayer’); replyElem.innerText ‘思考中…’; inputElem.disabled true; try { const response await fetch(‘http://localhost:8000/chat’, { method: ‘POST’, headers: { ‘Content-Type’: ‘application/json’ }, body: JSON.stringify({ text: text, session_id: ‘user_01’ }) }); const result await response.json(); if (result.success) { replyElem.innerText result.reply_text; // 如果有音频 URL则播放 if (result.digital_human result.digital_human.audio_url) { audioElem.src result.digital_human.audio_url; audioElem.style.display ‘block’; audioElem.play(); } // 此处可根据 result.digital_human.animation_data 更新数字人动画 console.log(‘动画数据:’, result.digital_human.animation_data); } else { replyElem.innerText ‘请求失败: ‘ (result.detail || ‘未知错误’); } } catch (error) { replyElem.innerText ‘网络或服务器错误: ‘ error.message; } finally { inputElem.disabled false; inputElem.value ‘’; inputElem.focus(); } } // 支持回车发送 document.getElementById(‘userInput’).addEventListener(‘keypress’, function(e) { if (e.key ‘Enter’) sendMessage(); }); /script /body /html4.4 数字人集成关键点与排错服务商选择与 SDK市面上数字人服务商众多如腾讯云智数、阿里云数字人、硅基、魔珐等。集成第一步是阅读官方文档获取 SDK 和 API 密钥。重点查看“语音合成TTS”、“驱动数据生成”和“形象渲染”三部分 API。驱动数据格式不同服务商提供的驱动数据格式不同可能是包含唇形、表情、肢体动作关键帧的 JSON 数据。前端需要对应的渲染引擎通常是 WebGL 或 Wasm来解析并驱动模型。音画同步这是一个工程难点。简单的方案是后端返回音频 URL 和每一句对应的动画数据时间戳前端播放音频时根据时间戳切换动画状态。复杂的方案需要服务端生成带音轨的完整视频流。性能与成本数字人渲染和 TTS 对计算资源和网络带宽要求较高。在原型阶段可以先专注于 Agent 逻辑用静态图片或简单动画代替真实数字人待核心流程跑通后再集成。常见集成问题问题排查方向SDK 初始化失败检查app_id、app_key是否正确检查网络连通性是否能访问服务商域名检查 SDK 版本与 Python 版本兼容性。合成成功但无音频/动画检查返回的数据结构确认audio_url是否有效、animation_data格式是否符合前端渲染引擎要求。用 Postman 直接调用 API 测试。前端无法播放音频检查音频 URL 是否可公开访问CORS 问题检查音频格式MP3、WAV浏览器是否支持。延迟过高数字人合成和 TTS 是耗时操作。考虑异步处理接口立即返回通过 WebSocket 或轮询通知前端结果。对实时性要求高的场景需要评估服务商的实时渲染 API。5. 项目三构建基于本地知识的 RAG 问答系统当 Agent 需要回答特定领域如公司内部文档、产品手册的问题时必须为其提供相关知识。RAG检索增强生成是当前最主流的技术方案。我们将使用 LangChain 和 ChromaDB一个轻量级向量数据库在本地搭建一个 RAG 问答系统。5.1 RAG 系统工作流程知识库加载与切分将 PDF、TXT、Markdown 等文档加载进来并按语义切分成片段Chunks。向量化与存储使用嵌入模型Embedding Model将文本片段转换为向量存入向量数据库。检索当用户提问时将问题也转换为向量在数据库中查找最相似的文本片段。增强生成将检索到的相关片段作为上下文与问题一起提交给 LLM让 LLM 生成答案。5.2 实现本地知识库构建# rag/knowledge_base.py import os from langchain_community.document_loaders import DirectoryLoader, TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.vectorstores import Chroma from langchain_community.embeddings import OllamaEmbeddings # 使用本地 Ollama 嵌入模型 # 或者使用 OpenAI 嵌入from langchain_openai import OpenAIEmbeddings class KnowledgeBase: def __init__(self, persist_directory“./chroma_db”): self.persist_directory persist_directory # 使用本地 Ollama 的 nomic-embed-text 模型无需 API 密钥 self.embeddings OllamaEmbeddings(model“nomic-embed-text”) # 或者使用 OpenAI: self.embeddings OpenAIEmbeddings(openai_api_keyos.getenv(“OPENAI_API_KEY”)) self.vectorstore None def load_and_split_documents(self, data_path“./data”): 从指定目录加载文档并切分。 # 加载所有 .txt 文件 loader DirectoryLoader(data_path, glob“**/*.txt”, loader_clsTextLoader) documents loader.load() print(f“已加载 {len(documents)} 个文档。”) # 文本切分器 text_splitter RecursiveCharacterTextSplitter( chunk_size500, # 每个片段约500字符 chunk_overlap50, # 片段间重叠50字符保持上下文连贯 separators[“\n\n”, “\n”, “。”, “”, “”, “,”, “ “, “”] ) splits text_splitter.split_documents(documents) print(f“切分为 {len(splits)} 个文本片段。”) return splits def create_vectorstore(self, splits): 创建向量存储并持久化。 self.vectorstore Chroma.from_documents( documentssplits, embeddingself.embeddings, persist_directoryself.persist_directory ) self.vectorstore.persist() print(f“向量数据库已创建并保存至 {self.persist_directory}”) def load_existing_vectorstore(self): 加载已存在的向量数据库。 if os.path.exists(self.persist_directory): self.vectorstore Chroma( persist_directoryself.persist_directory, embedding_functionself.embeddings ) print(“已加载现有向量数据库。”) return True else: print(“未找到已有的向量数据库请先创建。”) return False def similarity_search(self, query: str, k3): 在知识库中检索与查询最相关的 k 个片段。 if not self.vectorstore: raise ValueError(“向量数据库未初始化请先创建或加载。”) docs self.vectorstore.similarity_search(query, kk) return docs # 使用示例 if __name__ “__main__”: kb KnowledgeBase() # 首次运行加载文档并构建向量库 # splits kb.load_and_split_documents() # kb.create_vectorstore(splits) # 后续运行直接加载 if not kb.load_existing_vectorstore(): # 如果没有则创建 splits kb.load_and_split_documents() kb.create_vectorstore(splits) # 测试检索 test_query “公司的年假政策是怎样的” results kb.similarity_search(test_query) for i, doc in enumerate(results): print(f“\n--- 结果 {i1} ---“) print(doc.page_content[:200]) # 打印前200个字符5.3 创建 RAG 增强的问答链我们将把检索到的文档作为上下文构造一个提示词模板让 LLM 基于此上下文回答问题。# rag/qa_chain.py from langchain.chains import RetrievalQA from langchain.prompts import PromptTemplate from langchain_openai import ChatOpenAI from knowledge_base import KnowledgeBase # 导入上面定义的类 import os def create_rag_agent(): # 1. 加载知识库 kb KnowledgeBase() if not kb.load_existing_vectorstore(): raise RuntimeError(“请先构建知识库向量数据库。”) # 2. 初始化 LLM llm ChatOpenAI( model“gpt-3.5-turbo”, temperature0, openai_api_keyos.getenv(“OPENAI_API_KEY”) ) # 3. 构建提示词模板 prompt_template “”“请根据以下上下文信息回答问题。如果上下文信息不足以回答问题请直接说‘根据现有资料无法回答该问题’不要编造信息。 上下文 {context} 问题{question} 请给出准确、简洁的回答”“” PROMPT PromptTemplate( templateprompt_template, input_variables[“context”, “question”] ) # 4. 创建检索问答链 qa_chain RetrievalQA.from_chain_type( llmllm, chain_type“stuff”, # 将检索到的所有文档“塞”进上下文 retrieverkb.vectorstore.as_retriever(search_kwargs{“k”: 3}), chain_type_kwargs{“prompt”: PROMPT}, return_source_documentsTrue # 返回源文档便于调试 ) return qa_chain if __name__ “__main__”: qa create_rag_agent() while True: query input(“\n请输入问题 (输入‘退出’结束): “) if query in [“退出”, “exit”]: break result qa.invoke({“query”: query}) print(f“\n回答{result[‘result’]}”) # 可选查看来源 # for doc in result[‘source_documents’]: # print(f“来源{doc.metadata.get(‘source’, ‘N/A’)} - {doc.page_content[:100]}…”)5.4 RAG 系统优化与排错一个基础的 RAG 系统很容易搭建但要使其效果好需要在以下环节下功夫文档预处理原始文档可能包含无关的页眉页脚、广告、乱码。加载前需要进行清洗。文本切分Chunking这是影响检索精度的关键。chunk_size太小会丢失上下文太大会引入噪声。对于中文按句号、问号等自然分隔符切分比单纯按字符数更有效。可以尝试ChineseRecursiveTextSplitter等针对中文的切分器。嵌入模型Embedding嵌入模型的质量直接决定检索的相关性。对于中文text2vec、BGE系列模型通常比通用模型效果更好。可以在 Hugging Face 上寻找并集成。检索策略除了简单的相似度搜索similarity_search还可以尝试MMR最大边际相关性来平衡相关性和多样性或者使用SelfQueryRetriever让 LLM 自动从问题中提取元数据过滤条件。提示工程提示词模板决定了 LLM 如何利用上下文。清晰的指令如“基于上下文回答不要编造”能显著减少幻觉Hallucination。常见 RAG 问题问题原因与解决方案检索不到相关内容1. 检查文档是否成功加载和切分。2. 检查向量数据库中的文档数量。3. 尝试调整chunk_size增大或减小。4. 更换更强的嵌入模型。回答与文档无关幻觉1. 强化提示词要求“严格基于上下文”。2. 检查检索到的片段是否真的相关打印source_documents。3. 增加检索数量k提供更多上下文。4. 使用chain_type“map_reduce”或“refine”处理长上下文但速度会慢。回答“根据现有资料无法回答”过于频繁1. 可能检索阈值太高或k值太小尝试增大k。2. 检查问题是否太模糊可以尝试在提示词中让模型进行追问。构建向量库速度慢1. 使用本地嵌入模型如 Ollama可避免网络延迟但需消耗本地 GPU/CPU。2. 对于大批量文档考虑分批处理并加入进度提示。6. 项目四设计商业级智能体——多工具协作与状态管理一个真正的商业级智能体往往需要协调多个工具处理复杂、多步骤的任务并能在长时间对话中保持状态记忆。我们将使用 LangGraphLangChain 的一个扩展库来构建一个具备复杂工作流的智能体。6.1 理解 LangGraph基于图的 Agent 编排LangGraph 允许你将 Agent 的工作流定义为一个有向图Graph其中节点Node可以是工具调用、LLM 推理或条件判断边Edge决定了执行流程。这非常适合需要循环、分支和多步规划的场景。6.2 构建一个旅行规划智能体假设我们的智能体需要帮用户规划一次旅行它需要1. 查询天气2. 查询航班3. 推荐景点。我们将模拟这些工具。# advanced_agent/travel_agent.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain.tools import tool from langchain_community.tools.tavily_search import TavilySearchResults import os # 1. 定义状态结构 class AgentState(TypedDict): “”“智能体的状态在整个工作流中传递。”“” user_request: str # 原始用户请求 plan: str # 生成的计划 weather_info: str # 天气信息 flight_info: str # 航班信息 attraction_info: str # 景点信息 next_step: str # 决定下一步做什么 # 2. 定义工具模拟 tool def get_weather(city: str) - str: “”“获取指定城市的天气预报。输入是城市名。”“” # 模拟返回 return f“{city} 未来三天天气晴朗气温 20-25°C适合出行。” tool def search_flights(departure: str, destination: str, date: str) - str: “”“查询航班信息。输入是出发地、目的地和日期。”“” # 模拟返回 return f“找到从 {departure} 到 {destination} 在 {date} 的航班航班号 CA1234时间 08:00-10:00价格 1200元。” tool def recommend_attractions(city: str) - str: “”“推荐城市的旅游景点。输入是城市名。”“” # 模拟返回 return f“{city} 的推荐景点西湖、雷峰塔、灵隐寺。” # 3. 初始化 LLM 和工具 llm ChatOpenAI(model“gpt-3.5-turbo”, temperature0, api_keyos.getenv(“OPENAI_API_KEY”)) tools [get_weather, search_flights, recommend_attractions] llm_with_tools llm.bind_tools(tools) # 4. 定义各个节点函数 def planner_node(state: AgentState) - AgentState: “”“规划节点分析用户请求决定需要哪些信息。”“” messages [ (“system”, “你是一个旅行规划助手。请分析用户请求决定需要查询天气、航班还是景点信息。输出 ‘weather’, ‘flight’, ‘attraction’ 中的一个或多个用逗号分隔。”), (“human”, state[“user_request”]) ] response llm_with_tools.invoke(messages) # 简单解析实际应更健壮 plan response.content.lower() return {“plan”: plan} def weather_node(state: AgentState) - AgentState: “”“天气查询节点。”“” if “weather” in state[“plan”]: # 简单从请求中提取城市实际应用可用 NER 模型 city “杭州” # 模拟提取 info get_weather.invoke({“city”: city}) return {“weather_info”: info} return {“weather_info”: “未查询天气”} def flight_node(state: AgentState) - AgentState: “”“航班查询节点。”“” if “flight” in state[“plan”]: info search_flights.invoke({“departure”: “北京”, “destination”: “杭州”, “date”: “2024-10-01”}) return {“flight_info”: info} return {“flight_info”: “未查询航班”} def attraction_node(state: AgentState) - AgentState: “”“景点推荐节点。”“” if “attraction” in state[“plan”]: city “杭州” info recommend_attractions.invoke({“city”: city}) return {“attraction_info”: info} return {“attraction_info”: “未推荐景点”} def synthesizer_node(state: AgentState) - AgentState: “”“汇总节点将所有信息整合成最终答复。”“” context f“”” 用户请求{state[‘user_request’]} 天气信息{state.get(‘weather_info’, ‘无’)} 航班信息{state.get(‘flight_info’, ‘无’)} 景点信息{state.get(‘attraction_info’, ‘无’)} “”” messages [ (“system”, “你是一个旅行规划助手。请根据以上收集的信息生成一份友好、详细的旅行计划回复给用户。”), (“human”, context) ] response llm.invoke(messages) return {“next_step”: “end”, “final_reply”: response.content} # 5. 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(“planner”, planner_node) workflow.add_node(“weather”, weather_node) workflow.add_node(“flight”, flight_node) workflow.add_node(“attraction”, attraction_node) workflow.add_node(“synthesizer”, synthesizer_node) # 设置边流程 workflow.set_entry_point(“planner”) # 规划后并行执行天气、航班、景点查询 workflow.add_edge(“planner”, “weather”) workflow.add_edge(“planner”, “flight”) workflow.add_edge(“planner”, “attraction”) # 所有信息收集完成后进入汇总节点 workflow.add_conditional_edges( “weather”, lambda x: “synthesizer” if x.get(“weather_info”) else END ) workflow.add_edge(“flight”, “synthesizer”) workflow.add_edge(“attraction”, “synthesizer”) workflow.add_edge(“synthesizer”, END) # 编译图 app workflow.compile() # 6. 运行智能体 if __name__ “__main__”: # 模拟用户请求 user_input “我想下个月去杭州旅游帮我看看天气和航班再推荐几个景点。” initial_state {“user_request”: user_input} # 运行图 final_state app.invoke(initial_state) print(“\n 旅行规划结果 “) print(final_state.get(“final_reply”, “未生成回复”)) print(“\n 中间状态 “) for key, value in final_state.items(): if key ! “final_reply”: print(f“{key}: {value}”)6.3 商业级考量记忆、持久化与监控上面的例子是一个简单的线性流程。真实的商业级智能体还需要对话记忆Memory让 Agent 记住之前的对话历史。LangChain 提供了多种 Memory 组件ConversationBufferMemory,ConversationSummaryMemory等可以集成到 Agent 或 Graph 的状态中。状态持久化对于长时间运行或服务重启需要将 Agent 的状态如对话历史、任务进度保存到数据库如 Redis、PostgreSQL。工具可靠性真实工具调用可能失败网络超时、API 限流。需要为每个工具调用添加重试、降级和超时机制。可观测性Observability记录每一次 LLM 调用、工具调用的输入输出、耗时和 Token 消耗。这对于调试、成本核算和性能优化至关重要。可以集成 LangSmith 或自定义日志。权限与安全不同工具可能对应不同权限级别的操作如查询内部数据 vs 发送邮件。需要在调用工具前进行权限校验。6.4 生产环境部署建议API 服务化使用 FastAPI 或 Django 将智能体封装成 RESTful API 或 WebSocket 服务。配置管理所有模型参数、工具开关、提示词模板都应外置为配置文件如 YAML便于不同环境开发、测试、生产切换。异步处理对于耗时任务如文档处理、复杂规划应采用异步任务队列如 Celery Redis避免阻塞 HTTP 请求。限流与熔断对 LLM API 和关键工具接口设置限流防止意外流量导致服务雪崩或产生高额费用。版本管理对智能体的提示词、工具集、工作流进行版本控制便于回滚和 A/B 测试。从简单的工具调用 Agent到结合数字人的交互前端再到拥有私有知识的 RAG 系统最后到用 LangGraph 编排的复杂工作流我们完成了一个 AI 智能体从概念到商业级设计的完整实践路径。每个环节都有其技术重点和工程考量真正的挑战不在于实现单一功能而在于如何让这些组件稳定、高效、安全地协同工作。建议你从一个最小可运行的案例开始逐步增加复杂度并在每个阶段都建立完善的测试和监控这样才能构建出真正可用的 AI 应用。