这类项目最值得关注的不是“100个”这个数字而是它试图解决一个非常实际的问题如何让开发者尤其是想转型或提升的开发者能真正上手大模型智能体Agent开发并形成可落地的项目经验。很多教程停留在概念和“Hello World”级别而企业级实战需要处理任务拆解、工具调用、状态管理、错误处理和系统集成等一系列复杂问题。如果你正从传统开发转向AI应用或者想系统掌握Agent开发这篇文章会帮你理清从环境搭建到项目实战的核心路径并告诉你哪些环节最容易踩坑。直接说结论Agent开发的核心不是调用一个API而是设计一个能理解目标、使用工具、处理异常并完成复杂流程的“智能工作流”。学完概念后最关键的一步是找到一个有清晰输入、明确输出和完整工具链的实战项目从头到尾跑通它。1. 先拆解“企业级实战”到底在考什么很多人看到“企业级”就觉得高深莫测其实拆开看就是比个人玩具项目多了几层要求。如果你打算靠这个技能求职或承接项目面试官或客户最关心的往往是下面这些点。1.1 不只是跑通Demo而是处理“脏数据”和“异常流”个人项目通常用清洗好的、格式完美的数据。企业环境里的数据可能是残缺的、格式混乱的或者来自多个不同系统的。你的Agent能否处理输入容错用户输入不完整、有错别字、意图模糊时Agent是直接报错还是能尝试追问、澄清或从上下文中推理工具调用稳定性调用的外部API可能超时、返回错误格式或限流。你的Agent有没有重试机制、降级策略或友好的错误反馈状态持久化一个复杂的任务比如生成一份报告需要先查数据、再分析、最后排版可能被中断。Agent能否记住当前进度支持断点续跑在实战中我建议第一个项目不要选太开放的如“做一个聊天机器人”而是选一个输入输出相对固定但内部逻辑需要多步判断和工具调用的。例如“根据用户提供的产品名称和模糊需求自动查询数据库、对比竞品参数并生成一份结构化的对比报告”。这个过程中你会遇到产品名称不准确、数据库查询无结果、竞品信息缺失等多种异常处理这些才是实战的开始。1.2 明确Agent的“行动边界”与“工具集”一个常见的误区是试图让一个Agent“包打天下”。成熟的架构里Agent更像一个调度中心它的核心能力是“决策”和“编排”而不是亲自完成所有具体工作。工具Tools是它的手脚你需要为Agent集成一系列工具函数比如search_web网络搜索、query_database查数据库、call_calculator计算、generate_chart画图等。在项目设计阶段就要明确列出这个Agent被允许使用哪些工具。规划Planning是它的大脑给定一个目标Agent需要能将其分解为子任务序列Plan并决定何时调用哪个工具。例如目标“写一篇关于Agent的博客”可能被分解为[搜索最新资料 - 整理大纲 - 撰写引言 - 撰写正文 - 生成配图建议 - 排版]。记忆Memory是它的经验短期记忆会话上下文让Agent能理解多轮对话长期记忆向量数据库可以让它记住历史交互的关键信息避免重复劳动。在你的第一个实战项目中我建议将工具集控制在3-5个并且都是你能完全掌控的比如调用一个本地函数或你熟悉的公开API。先让Agent学会在有限的“武器库”里解决问题。1.3 从单智能体到多智能体协作与工作流简单任务一个Agent就够了。复杂业务流程如客服系统先由“理解Agent”分类问题再路由给“查询Agent”或“工单Agent”最后“回复Agent”合成答案则需要多个Agent协作这就是智能体工作流。编排Orchestration 如何定义Agent之间的触发关系、数据传递和依赖是串行、并行还是有条件分支通信Communication Agent之间如何交换信息是通过共享状态、消息队列还是直接函数调用共识与冲突解决 当多个Agent对同一问题有不同输出时如何裁决对于初学者不要一开始就挑战多智能体。你的学习路径应该是单智能体带工具 - 单智能体多步骤工作流 - 多智能体简单协作。目前主流的框架如LangChain的LangGraph、微软的AutoGen、以及项目里可能提到的AgentScope等都提供了构建工作流的能力。先理解它们的基本抽象图、节点、边、状态再用一个简单的两Agent协作场景比如一个负责检索一个负责总结来实践。2. 搭建你的第一个可运行Agent开发环境理论再好不如跑一行代码。下面是一个兼顾学习和轻度实战的本地环境搭建方案它不追求最新最全但求稳定、可复现能支撑你完成大多数入门和中级项目。2.1 核心三件套Python环境、大模型API、开发框架1. Python环境管理使用Conda这是避免依赖地狱的第一步。不要用系统Python。# 创建并激活一个专用于Agent开发的虚拟环境 conda create -n agent_dev python3.10 -y conda activate agent_dev2. 大模型接入选择一种国内推荐DeepSeekAgent需要一个“大脑”来理解指令和做决策。你有两个主流选择云端API推荐起步 方便快捷无需强大显卡。国内可以直接使用DeepSeek、智谱GLM、月之暗面Kimi等提供的API。国外则是OpenAI的GPT系列、Anthropic的Claude等。操作去对应平台注册获取API Key。通常有免费额度供学习。优势模型能力强更新快省去部署烦恼。注意API调用有网络延迟和成本且你的数据会发送到厂商服务器敏感数据需注意。本地部署模型 数据隐私性好无网络依赖。可使用Ollama、vLLM、Transformers等工具部署较小的开源模型如Qwen2.5-7B、Llama 3.1-8B。操作对硬件尤其是GPU显存有要求。7B模型通常需要8GB以上显存才能流畅运行。优势完全可控适合处理内部数据。注意本地小模型的推理和规划能力通常弱于顶级云端大模型可能需要更多提示工程Prompt Engineering。对于绝大多数学习者我强烈建议从云端API开始先把精力集中在Agent逻辑本身而不是模型部署和调优上。这里以DeepSeek API为例因为它对中文支持好且免费额度充足。# 安装必要的库 pip install openai # DeepSeek API兼容OpenAI SDK格式3. 开发框架选择LangChain或直接SDKLangChain/LangGraph 当前最流行的Agent开发框架之一。它提供了Chain、Agent、Tool、Memory等高级抽象能快速搭建原型。LangGraph专门用于构建有状态、多步骤的工作流。学习曲线稍陡但生态丰富。pip install langchain langchain-openai langgraph直接使用大模型SDK 如果你喜欢更底层的控制可以直接用OpenAI SDK或各厂商的SDK自己实现Agent的循环逻辑。这更灵活但需要写更多样板代码。其他框架 项目中可能提到的AgentScope、AutoGen等各有侧重。建议先精通一个如LangChain再触类旁通。2.2 验证环境创建一个最简单的“计算器Agent”让我们用5分钟验证环境是否工作并理解Agent的基本构成。这个Agent只做一件事判断用户问题是否需要计算如果需要就调用一个计算工具。# 文件simple_calculator_agent.py import os from langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate from langchain.tools import tool # 1. 设置你的大模型这里以DeepSeek为例需设置api_base os.environ[OPENAI_API_KEY] your-deepseek-api-key # 替换成你的真实Key llm ChatOpenAI( modeldeepseek-chat, base_urlhttps://api.deepseek.com/v1, # DeepSeek的API端点 temperature0 ) # 2. 定义一个工具计算器 tool def calculator(expression: str) - str: 计算一个数学表达式的结果。支持加减乘除和括号。例如(3 5) * 2 try: # 警告实际生产环境请使用更安全的计算方式如ast.literal_eval或专用库这里仅为演示。 result eval(expression) return f计算结果为{result} except Exception as e: return f计算错误{e} # 3. 准备工具列表和提示词模板 tools [calculator] prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的助手可以回答一般问题并且拥有一个计算器工具。如果用户的问题涉及数学计算请务必使用计算器工具来获得精确答案。), (placeholder, {chat_history}), (human, {input}), (placeholder, {agent_scratchpad}), ]) # 4. 创建Agent agent create_tool_calling_agent(llmllm, toolstools, promptprompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) # verboseTrue 会打印思考过程 # 5. 运行Agent if __name__ __main__: # 测试用例1需要计算 result1 agent_executor.invoke({input: 请问 (12 34) * 5.6 等于多少}) print(用户问题请问 (12 34) * 5.6 等于多少) print(fAgent回答{result1[output]}\n) # 测试用例2普通问答 result2 agent_executor.invoke({input: 你是谁}) print(用户问题你是谁) print(fAgent回答{result2[output]}\n) # 测试用例3复杂计算 result3 agent_executor.invoke({input: 计算圆的面积如果半径是7.5厘米。}) print(用户问题计算圆的面积如果半径是7.5厘米。) print(fAgent回答{result3[output]})运行并观察python simple_calculator_agent.py如果一切正常你会看到类似以下的输出其中包含了Agent的“思考过程”因为它设置了verboseTrue Entering new AgentExecutor chain... 用户需要计算一个数学表达式。我应该使用计算器工具。 Action: calculator Action Input: (12 34) * 5.6 Observation: 计算结果为257.6 Thought: 我已经得到了计算结果可以回答用户了。 Final Answer: (12 34) * 5.6 的计算结果是 257.6。 用户问题请问 (12 34) * 5.6 等于多少 Agent回答: (12 34) * 5.6 的计算结果是 257.6。这个简单的例子包含了Agent的核心要素LLM大脑、Prompt指令、Tools工具、Executor执行器。看到这个输出你的基础环境就过关了。2.3 常见环境踩坑点API Key 错误或网络问题 最常见的错误。确保Key正确、有余额并且网络能访问API端点。国内调用国外API可能需要配置网络环境但严禁讨论任何相关工具和方法请使用国内合规的模型服务。依赖版本冲突 LangChain等框架更新很快。如果遇到奇怪报错先检查pip list看关键包langchain,langchain-openai,openai等版本是否兼容。建议使用项目要求的特定版本或锁定较新的稳定版本。本地模型部署失败 如果选择Ollama确保下载的模型名称正确且显存足够。7B模型通常需要至少8GB GPU显存。可以通过ollama run qwen2.5:7b测试模型是否能正常对话。工具定义错误tool装饰的函数必须有清晰的文档字符串docstringAgent依赖它来决定是否调用该工具。输入参数的类型提示如expression: str也很重要。3. 从Demo到项目设计并实现一个“智能数据查询助手”现在我们来做一个更接近实战的小项目。目标构建一个Agent它能理解用户用自然语言提出的数据查询需求自动将其转换为SQL语句查询数据库并将结果以自然语言总结出来。3.1 项目定义与工具准备项目目标用户问“上个月销售额最高的产品是什么”Agent应能理解“上个月”是时间范围“销售额最高”是排序和聚合然后生成SQLSELECT product_name, SUM(sales) FROM orders WHERE order_date ‘2024-04-01’ GROUP BY product_name ORDER BY SUM(sales) DESC LIMIT 1执行查询并返回“上个月销售额最高的产品是XXX总销售额为YYY元。”我们需要三个核心工具数据库连接与查询工具 接收SQL返回查询结果。数据库Schema查看工具 让Agent知道数据库里有哪些表、字段及其含义否则它无法生成正确的SQL。日期处理工具可选但重要 将“上个月”、“本周”等自然语言转换为具体的日期范围。我们先模拟一个简单的SQLite数据库和几张表。# 文件setup_database.py import sqlite3 import pandas as pd from datetime import datetime, timedelta # 创建并连接数据库 conn sqlite3.connect(sales_data.db) cursor conn.cursor() # 创建产品表 cursor.execute( CREATE TABLE IF NOT EXISTS products ( product_id INTEGER PRIMARY KEY, product_name TEXT NOT NULL, category TEXT ) ) # 创建订单表 cursor.execute( CREATE TABLE IF NOT EXISTS orders ( order_id INTEGER PRIMARY KEY, product_id INTEGER, sales REAL, order_date DATE, FOREIGN KEY (product_id) REFERENCES products (product_id) ) ) # 插入模拟数据 products_data [(1, 智能手机X, 电子产品), (2, 蓝牙耳机Y, 电子产品), (3, 咖啡机Z, 家电), (4, 办公椅A, 家具)] cursor.executemany(INSERT OR IGNORE INTO products VALUES (?,?,?), products_data) # 生成最近3个月的模拟订单数据 orders_data [] start_date datetime.now() - timedelta(days90) for i in range(1, 101): product_id (i % 4) 1 sales round(100 (i * 10) % 500, 2) order_date (start_date timedelta(days(i % 90))).strftime(%Y-%m-%d) orders_data.append((i, product_id, sales, order_date)) cursor.executemany(INSERT OR IGNORE INTO orders VALUES (?,?,?,?), orders_data) conn.commit() print(数据库‘sales_data.db’及模拟数据已创建。) print(产品表样例) print(pd.read_sql_query(SELECT * FROM products LIMIT 5, conn)) print(\n订单表样例) print(pd.read_sql_query(SELECT * FROM orders LIMIT 5, conn)) conn.close()运行这个脚本创建你的测试数据库。3.2 构建智能体集成工具与设计提示词现在我们来编写核心的Agent。我们将使用LangChain的create_sql_agent这是一个高级封装它内部已经集成了生成SQL、执行、纠错等逻辑非常适合这个场景。# 文件sql_query_agent.py import os from langchain_openai import ChatOpenAI from langchain_community.utilities import SQLDatabase from langchain_community.agent_toolkits import create_sql_agent from langchain_community.tools import QuerySQLDataBaseTool from datetime import datetime, timedelta # 1. 设置大模型同上 os.environ[OPENAI_API_KEY] your-deepseek-api-key llm ChatOpenAI(modeldeepseek-chat, base_urlhttps://api.deepseek.com/v1, temperature0) # 2. 连接数据库 db SQLDatabase.from_uri(sqlite:///sales_data.db) # 3. 创建SQL Agent agent_executor create_sql_agent( llmllm, dbdb, agent_typeopenai-tools, # 使用工具调用类型的Agent verboseTrue, # 以下是增强Agent能力的额外提示词 prefix你是一个专业的数据库分析师。你的任务是根据用户的问题查询数据库并给出清晰、准确的答案。 请遵循以下规则 1. 首先理解用户问题中的时间范围。如果用户提到“今天”、“昨天”、“本周”、“本月”、“上个月”等你需要将其转换为具体的SQL日期条件。 - 例如‘今天’ 指的是 ‘{today_date}’。 - ‘上个月’ 指的是 从 ‘{last_month_start}’ 到 ‘{last_month_end}’。 我会在系统提示词中为你动态更新这些日期 2. 在生成SQL前先查看相关的数据库表结构schema确保你引用的表名和字段名是正确的。 3. 生成的SQL语句必须语法正确并且只查询必要的数据。 4. 如果查询结果为空或出错请分析原因并尝试调整查询条件或告知用户可能的原因。 5. 最终答案应该用自然语言总结查询结果不要直接输出原始的SQL结果集。 ) # 4. 动态生成包含今日日期的提示词 def get_dynamic_prefix(): today datetime.now().strftime(%Y-%m-%d) last_month_start (datetime.now().replace(day1) - timedelta(days1)).replace(day1).strftime(%Y-%m-%d) last_month_end (datetime.now().replace(day1) - timedelta(days1)).strftime(%Y-%m-%d) prefix f你是一个专业的数据库分析师。你的任务是根据用户的问题查询数据库并给出清晰、准确的答案。 请遵循以下规则 1. 首先理解用户问题中的时间范围。如果用户提到“今天”、“昨天”、“本周”、“本月”、“上个月”等你需要将其转换为具体的SQL日期条件。 - 例如‘今天’ 指的是 ‘{today}’。 - ‘上个月’ 指的是 从 ‘{last_month_start}’ 到 ‘{last_month_end}’。 2. 在生成SQL前先查看相关的数据库表结构schema确保你引用的表名和字段名是正确的。 3. 生成的SQL语句必须语法正确并且只查询必要的数据。 4. 如果查询结果为空或出错请分析原因并尝试调整查询条件或告知用户可能的原因。 5. 最终答案应该用自然语言总结查询结果不要直接输出原始的SQL结果集。 return prefix # 由于create_sql_agent的prefix在创建时固定我们可以通过更新agent_executor.agent.llm_chain.prompt的方式动态修改这里为了简化我们先使用一个静态前缀。 # 在实际更复杂的场景中你可能需要自定义一个更灵活的Agent。 if __name__ __main__: # 测试几个查询 queries [ 总共有多少条订单记录, 上个月销售额最高的产品是什么, 咖啡机Z这个月的总销售额是多少, 哪个产品类别的总销售额最高, ] for query in queries: print(f\n{*50}) print(f用户问题{query}) print(f{*50}) try: # 注意由于我们使用了静态前缀关于“上个月”的日期可能不准确这正是一个可以改进的点。 result agent_executor.invoke({input: query}) print(fAgent回答{result[output]}) except Exception as e: print(f执行出错{e})运行这个Agentpython sql_query_agent.py观察输出。一个成熟的SQL Agent会先查看数据库Schema然后生成SQL执行并返回总结。你可能会看到它生成的SQL类似SELECT p.product_name, SUM(o.sales) as total_sales FROM orders o JOIN products p ON o.product_id p.product_id WHERE o.order_date BETWEEN ‘2024-04-01’ AND ‘2024-04-30’ GROUP BY p.product_name ORDER BY total_sales DESC LIMIT 13.3 项目深化错误处理与能力扩展一个基础版本跑通后就要考虑企业级需求了SQL注入与安全性 我们的Agent直接执行模型生成的SQL。这在生产环境是极度危险的必须对生成的SQL进行严格的校验和限制例如只允许SELECT禁止DROP、DELETE等。可以使用sqlparse库进行解析或使用数据库用户的只读权限。复杂日期/语义处理 我们的简单前缀提示可能不够。更好的做法是专门设计一个“时间解析工具”。这个工具接收“上个月”、“本周三”等字符串返回标准的日期范围或SQL WHERE条件片段然后让Agent在生成SQL时调用这个工具的结果。查询优化与解释 对于复杂的查询Agent可以尝试解释它为什么这么写或者当查询很慢时建议添加索引。结果可视化 可以集成一个工具当用户问“画一个趋势图”时调用matplotlib或plotly生成图表并保存为文件然后将文件路径或图片描述返回给用户。这个“智能数据查询助手”项目麻雀虽小五脏俱全。它涉及了自然语言理解、工具调用SQL执行、状态管理日期上下文、错误处理SQL错误。把它做扎实你对Agent开发的理解会上一个大台阶。4. 进阶实战构建多智能体工作流处理复杂任务单智能体处理线性任务够了但面对“调研一个主题并撰写报告”这类需要检索、分析、写作、审核多步骤的任务就需要工作流了。我们使用LangGraph来构建一个简单的“研究助手”工作流。场景用户输入一个研究主题如“大模型智能体的最新发展趋势”工作流自动执行1. 网络搜索 - 2. 内容总结 - 3. 报告撰写 - 4. 报告评分。如果评分过低则重新搜索或总结。4.1 定义工作流中的节点智能体我们将定义四个节点每个节点可以看作一个具备特定能力的智能体或一个函数。# 文件research_workflow.py import os from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI from langchain_community.tools import TavilySearchResults from langchain_core.messages import HumanMessage, SystemMessage import operator # 0. 设置环境和模型 os.environ[OPENAI_API_KEY] your-openai-or-other-api-key os.environ[TAVILY_API_KEY] your-tavily-api-key # 用于搜索可在Tavily官网申请免费额度 llm ChatOpenAI(modelgpt-4o-mini, temperature0.7) # 使用一个适合创意和总结的模型 # 1. 定义工作流的状态State class ResearchState(TypedDict): # 消息历史 messages: Annotated[List, add_messages] # 研究主题 topic: str # 搜索到的原始内容 raw_search_results: str # 总结后的内容 summarized_content: str # 生成的报告 report: str # 报告评分 (1-10) report_score: int # 2. 定义各个节点函数 def search_node(state: ResearchState) - ResearchState: 节点1搜索。使用Tavily搜索工具获取最新信息。 print(f[Search Node] 正在搜索主题: {state[topic]}) search_tool TavilySearchResults(max_results3) # 限制3条结果 search_query f{state[topic]} latest developments 2024 try: results search_tool.invoke(search_query) # 将结果拼接成字符串 combined_results \n---\n.join([f来源{r.get(title, N/A)}\n内容{r.get(content, N/A)} for r in results]) return {raw_search_results: combined_results} except Exception as e: print(f搜索失败: {e}) return {raw_search_results: f搜索时出错: {e}} def summarize_node(state: ResearchState) - ResearchState: 节点2总结。基于搜索到的原始内容生成一份简洁的摘要。 print(f[Summarize Node] 正在总结搜索内容...) if not state[raw_search_results] or 出错 in state[raw_search_results]: return {summarized_content: 无有效搜索内容可供总结。} prompt f 你是一个专业的研究助理。请根据以下搜索到的原始材料提炼出关于“{state[topic]}”的核心观点、关键事实和主要趋势。 摘要要求简洁、客观、信息密度高字数在300字以内。 原始材料 {state[raw_search_results]} 摘要 messages [SystemMessage(content你是一个专业的总结者。), HumanMessage(contentprompt)] response llm.invoke(messages) return {summarized_content: response.content} def write_report_node(state: ResearchState) - ResearchState: 节点3撰写报告。基于总结的内容撰写一份结构完整的简短报告。 print(f[Write Report Node] 正在撰写报告...) if not state[summarized_content]: return {report: 无法撰写报告因为缺少总结内容。} prompt f 你是一名技术作家。请基于以下关于“{state[topic]}”的摘要撰写一份结构清晰、内容充实的简短报告。 报告应包含引言、主要发展趋势、当前面临的挑战、未来展望等部分。总字数控制在500字左右。 摘要内容 {state[summarized_content]} 报告 messages [SystemMessage(content你是一名优秀的报告撰写者。), HumanMessage(contentprompt)] response llm.invoke(messages) return {report: response.content} def evaluate_report_node(state: ResearchState) - ResearchState: 节点4评估报告。对生成的报告进行质量评分并决定工作流是否结束。 print(f[Evaluate Node] 正在评估报告质量...) if not state[report]: return {report_score: 0} prompt f 你是一名严格的评审。请对以下关于“{state[topic]}”的技术报告进行评分1-10分10分为最高。 评分标准 - 结构清晰度 (2分) - 信息准确性与相关性 (3分) - 内容深度与洞察力 (3分) - 语言流畅度 (2分) 请只输出一个整数分数。 报告内容 {state[report][:1000]} # 限制长度以防token超限 分数 messages [SystemMessage(content你是一名公正的评审。), HumanMessage(contentprompt)] response llm.invoke(messages) try: score int(response.content.strip()) except: score 5 # 解析失败则给中间分 print(f 评估得分: {score}/10) # 将分数存入状态 return {report_score: score} def should_continue(state: ResearchState) - str: 条件判断函数根据评分决定下一步是重新搜索还是结束。 score state.get(report_score, 0) # 如果评分低于6分我们让工作流跳回搜索节点重新开始否则结束。 # 注意这是一个简单演示实际中可能设置最大循环次数避免死循环。 if score 6: print(f 报告评分{score}分过低将重新开始研究流程。) return search # 返回下一个要执行的节点名 else: print(f 报告评分{score}分合格工作流结束。) return END # 3. 构建工作流图 workflow StateGraph(ResearchState) # 添加节点 workflow.add_node(search, search_node) workflow.add_node(summarize, summarize_node) workflow.add_node(write_report, write_report_node) workflow.add_node(evaluate, evaluate_report_node) # 设置边的起点 workflow.set_entry_point(search) # 添加条件边 workflow.add_conditional_edges( evaluate, # 从评估节点出发 should_continue, # 根据这个函数的返回值决定去哪 { search: search, # 如果返回search则跳回搜索节点 END: END # 如果返回END则结束 } ) # 添加普通边线性执行部分 workflow.add_edge(search, summarize) workflow.add_edge(summarize, write_report) workflow.add_edge(write_report, evaluate) # 编译图 app workflow.compile() # 4. 运行工作流 if __name__ __main__: # 初始化状态 initial_state: ResearchState { messages: [], topic: 大模型智能体AI Agent的最新发展趋势, raw_search_results: , summarized_content: , report: , report_score: 0 } print(开始执行研究助手工作流...) print(f研究主题: {initial_state[topic]}) print(- * 50) # 执行图 final_state app.invoke(initial_state, config{recursion_limit: 3}) # 限制递归深度为3次防止无限循环 print(\n *50) print(工作流执行完毕) print(f最终报告评分: {final_state.get(report_score)}/10) print(\n生成的报告内容预览) print(final_state.get(report, )[:500] ...) # 打印前500字符4.2 理解工作流的关键设计这个例子虽然简单但体现了多智能体工作流的核心思想状态State共享 我们定义了一个ResearchState字典所有节点都读写这个状态。这是智能体之间通信的主要方式。节点Node专精 每个节点只做一件事搜索、总结、写作、评估符合单一职责原则易于调试和替换。条件路由Conditional Edgeshould_continue函数根据报告评分动态决定工作流是结束还是重新开始。这赋予了工作流“循环”和“自我修正”的能力。图Graph结构 工作流被明确定义为一张图节点是处理单元边是执行顺序或条件逻辑。这比写一堆if-else嵌套要清晰和强大得多。要运行这个工作流你需要一个可用的OpenAI格式的API示例中用了gpt-4o-mini你可以换成其他模型。一个 Tavily 的API Key用于搜索它有免费额度。你也可以替换成其他搜索工具如Serper API、Google Search API等。4.3 企业级工作流的考量在实际项目中你还需要考虑错误处理与重试 任何一个节点如搜索API调用失败都可能出错。需要在节点内部或图层面添加重试逻辑和错误处理边。异步与并行 如果“搜索”和“查阅内部文档”可以同时进行就应该设计成并行节点以提高效率。LangGraph支持异步和并行执行。人工审核节点 在关键节点如报告发布前插入一个“人工审核”节点等待人工确认后再继续。持久化与可观测性 将每次运行的状态、中间结果和最终输出保存到数据库方便追溯和调试。使用langgraph的检查点Checkpoint功能可以实现状态持久化和断点续跑。5. 项目复盘与面试准备从会做到会讲完成几个项目后如何把它们变成你的竞争力面试官不会只看你的代码他更想听你为什么这么设计遇到了什么问题以及如何解决的。5.1 为你的项目准备“叙事线”针对任何一个你完成的Agent项目准备好回答以下问题项目背景与目标 一句话说清楚这个Agent是干什么的解决了什么痛点。例如“我开发了一个智能数据查询助手让业务人员能用自然语言直接查询数据库无需学习SQL。”架构与核心组件大脑LLM 你选用哪个模型为什么考虑成本、性能、中文能力、API稳定性。工具Tools 你为Agent配备了哪些工具为什么选这些如何保证工具调用的安全性和稳定性例如SQL注入防护、API限流与重试。记忆Memory 你用了哪种记忆会话记忆ConversationBufferMemory还是长期记忆向量数据库如何管理记忆的token消耗编排Orchestration 是单智能体还是工作流如果是工作流图是怎么设计的节点间如何传递数据关键挑战与解决方案提示工程Prompt Engineering 如何设计System Prompt和Few-shot Examples来让Agent更听话、更准确工具调用可靠性 如何处理工具调用失败网络超时、API错误有没有实现fallback机制成本与延迟优化 如何减少不必要的LLM调用或Token消耗例如缓存、思维链压缩、更精准的工具选择。评估与迭代 如何评估你的Agent效果有没有建立测试集如何根据反馈迭代Prompt和工具部署与监控如何部署你的AgentFastAPI封装成服务Streamlit做界面如何监控它的运行状态日志记录、关键指标如调用次数、成功率、平均响应时间。5.2 针对Agent开发的常见面试题除了项目本身你还需要了解一些通用知识Agent vs Chain vs Pipeline 说说你对它们的理解。核心区别在于Agent有决策能力决定下一步做什么调用哪个工具而Chain是固定的序列Pipeline可能更偏向数据流。ReAct vs Plan-and-Execute 两种主流的Agent推理框架。ReActReasoning Acting是边想边做每一步思考后决定行动Plan-and-Execute是先制定完整计划再执行。前者灵活后者可能更高效。你项目中用的是哪种思路LangChain, LlamaIndex, AutoGen等框架的优劣 不必全部精通但要对最常用的如LangChain有深入理解并了解其他框架的侧重点如LlamaIndex擅长检索AutoGen擅长多智能体对话。大模型幻觉Hallucination与缓解 Agent严重依赖LLM如何应对LLM胡说八道常用方法有要求提供引用来源检索增强生成RAG、设置验证步骤让另一个Agent或规则检查输出、在Prompt中强调“不知道就说不知道”。安全性考量 除了前面提到的SQL注入还要考虑工具权限控制不能让Agent随意调用删除文件的工具、用户输入过滤、输出内容安全审查、API密钥管理等。5.3 学习路线与资源建议回到标题中的“100个项目”其价值在于提供广泛的场景。但你的学习不应是线性的。更高效的方式是掌握核心范式 通过2-3个不同类型的项目如带工具调用的单智能体、基于RAG的问答Agent、多智能体工作流彻底吃透Agent的核心概念和主流框架LangChain/LangGraph的使用。垂直深入 选择一个你感兴趣的垂直领域如智能客服、数据分析、自动化办公、游戏NPC用Agent技术去解决该领域的一个具体问题并做深做透。关注底层与前沿 了解一些底层框架如DSPy、MCP和前沿架构如CrewAI、Microsoft Autogen理解其设计思想。关注开源模型如DeepSeek-V2、Qwen2.5的Agent能力进展。工程化实践 学习如何将你的Agent项目容器化Docker、编写测试用例、配置CI/CD、设计监控告警。这是从“玩具”到“产品”的关键一步。记住“100个项目”的目标不是刷数量而是通过足够的练习让你在面对一个全新的、模糊的需求时能快速将其分解为Agent可解决的子问题并选择合适的技术组件进行搭建。这才是企业真正需要的能力。 大模型智能体开发实战：从环境搭建到企业级项目落地
这类项目最值得关注的不是“100个”这个数字而是它试图解决一个非常实际的问题如何让开发者尤其是想转型或提升的开发者能真正上手大模型智能体Agent开发并形成可落地的项目经验。很多教程停留在概念和“Hello World”级别而企业级实战需要处理任务拆解、工具调用、状态管理、错误处理和系统集成等一系列复杂问题。如果你正从传统开发转向AI应用或者想系统掌握Agent开发这篇文章会帮你理清从环境搭建到项目实战的核心路径并告诉你哪些环节最容易踩坑。直接说结论Agent开发的核心不是调用一个API而是设计一个能理解目标、使用工具、处理异常并完成复杂流程的“智能工作流”。学完概念后最关键的一步是找到一个有清晰输入、明确输出和完整工具链的实战项目从头到尾跑通它。1. 先拆解“企业级实战”到底在考什么很多人看到“企业级”就觉得高深莫测其实拆开看就是比个人玩具项目多了几层要求。如果你打算靠这个技能求职或承接项目面试官或客户最关心的往往是下面这些点。1.1 不只是跑通Demo而是处理“脏数据”和“异常流”个人项目通常用清洗好的、格式完美的数据。企业环境里的数据可能是残缺的、格式混乱的或者来自多个不同系统的。你的Agent能否处理输入容错用户输入不完整、有错别字、意图模糊时Agent是直接报错还是能尝试追问、澄清或从上下文中推理工具调用稳定性调用的外部API可能超时、返回错误格式或限流。你的Agent有没有重试机制、降级策略或友好的错误反馈状态持久化一个复杂的任务比如生成一份报告需要先查数据、再分析、最后排版可能被中断。Agent能否记住当前进度支持断点续跑在实战中我建议第一个项目不要选太开放的如“做一个聊天机器人”而是选一个输入输出相对固定但内部逻辑需要多步判断和工具调用的。例如“根据用户提供的产品名称和模糊需求自动查询数据库、对比竞品参数并生成一份结构化的对比报告”。这个过程中你会遇到产品名称不准确、数据库查询无结果、竞品信息缺失等多种异常处理这些才是实战的开始。1.2 明确Agent的“行动边界”与“工具集”一个常见的误区是试图让一个Agent“包打天下”。成熟的架构里Agent更像一个调度中心它的核心能力是“决策”和“编排”而不是亲自完成所有具体工作。工具Tools是它的手脚你需要为Agent集成一系列工具函数比如search_web网络搜索、query_database查数据库、call_calculator计算、generate_chart画图等。在项目设计阶段就要明确列出这个Agent被允许使用哪些工具。规划Planning是它的大脑给定一个目标Agent需要能将其分解为子任务序列Plan并决定何时调用哪个工具。例如目标“写一篇关于Agent的博客”可能被分解为[搜索最新资料 - 整理大纲 - 撰写引言 - 撰写正文 - 生成配图建议 - 排版]。记忆Memory是它的经验短期记忆会话上下文让Agent能理解多轮对话长期记忆向量数据库可以让它记住历史交互的关键信息避免重复劳动。在你的第一个实战项目中我建议将工具集控制在3-5个并且都是你能完全掌控的比如调用一个本地函数或你熟悉的公开API。先让Agent学会在有限的“武器库”里解决问题。1.3 从单智能体到多智能体协作与工作流简单任务一个Agent就够了。复杂业务流程如客服系统先由“理解Agent”分类问题再路由给“查询Agent”或“工单Agent”最后“回复Agent”合成答案则需要多个Agent协作这就是智能体工作流。编排Orchestration 如何定义Agent之间的触发关系、数据传递和依赖是串行、并行还是有条件分支通信Communication Agent之间如何交换信息是通过共享状态、消息队列还是直接函数调用共识与冲突解决 当多个Agent对同一问题有不同输出时如何裁决对于初学者不要一开始就挑战多智能体。你的学习路径应该是单智能体带工具 - 单智能体多步骤工作流 - 多智能体简单协作。目前主流的框架如LangChain的LangGraph、微软的AutoGen、以及项目里可能提到的AgentScope等都提供了构建工作流的能力。先理解它们的基本抽象图、节点、边、状态再用一个简单的两Agent协作场景比如一个负责检索一个负责总结来实践。2. 搭建你的第一个可运行Agent开发环境理论再好不如跑一行代码。下面是一个兼顾学习和轻度实战的本地环境搭建方案它不追求最新最全但求稳定、可复现能支撑你完成大多数入门和中级项目。2.1 核心三件套Python环境、大模型API、开发框架1. Python环境管理使用Conda这是避免依赖地狱的第一步。不要用系统Python。# 创建并激活一个专用于Agent开发的虚拟环境 conda create -n agent_dev python3.10 -y conda activate agent_dev2. 大模型接入选择一种国内推荐DeepSeekAgent需要一个“大脑”来理解指令和做决策。你有两个主流选择云端API推荐起步 方便快捷无需强大显卡。国内可以直接使用DeepSeek、智谱GLM、月之暗面Kimi等提供的API。国外则是OpenAI的GPT系列、Anthropic的Claude等。操作去对应平台注册获取API Key。通常有免费额度供学习。优势模型能力强更新快省去部署烦恼。注意API调用有网络延迟和成本且你的数据会发送到厂商服务器敏感数据需注意。本地部署模型 数据隐私性好无网络依赖。可使用Ollama、vLLM、Transformers等工具部署较小的开源模型如Qwen2.5-7B、Llama 3.1-8B。操作对硬件尤其是GPU显存有要求。7B模型通常需要8GB以上显存才能流畅运行。优势完全可控适合处理内部数据。注意本地小模型的推理和规划能力通常弱于顶级云端大模型可能需要更多提示工程Prompt Engineering。对于绝大多数学习者我强烈建议从云端API开始先把精力集中在Agent逻辑本身而不是模型部署和调优上。这里以DeepSeek API为例因为它对中文支持好且免费额度充足。# 安装必要的库 pip install openai # DeepSeek API兼容OpenAI SDK格式3. 开发框架选择LangChain或直接SDKLangChain/LangGraph 当前最流行的Agent开发框架之一。它提供了Chain、Agent、Tool、Memory等高级抽象能快速搭建原型。LangGraph专门用于构建有状态、多步骤的工作流。学习曲线稍陡但生态丰富。pip install langchain langchain-openai langgraph直接使用大模型SDK 如果你喜欢更底层的控制可以直接用OpenAI SDK或各厂商的SDK自己实现Agent的循环逻辑。这更灵活但需要写更多样板代码。其他框架 项目中可能提到的AgentScope、AutoGen等各有侧重。建议先精通一个如LangChain再触类旁通。2.2 验证环境创建一个最简单的“计算器Agent”让我们用5分钟验证环境是否工作并理解Agent的基本构成。这个Agent只做一件事判断用户问题是否需要计算如果需要就调用一个计算工具。# 文件simple_calculator_agent.py import os from langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate from langchain.tools import tool # 1. 设置你的大模型这里以DeepSeek为例需设置api_base os.environ[OPENAI_API_KEY] your-deepseek-api-key # 替换成你的真实Key llm ChatOpenAI( modeldeepseek-chat, base_urlhttps://api.deepseek.com/v1, # DeepSeek的API端点 temperature0 ) # 2. 定义一个工具计算器 tool def calculator(expression: str) - str: 计算一个数学表达式的结果。支持加减乘除和括号。例如(3 5) * 2 try: # 警告实际生产环境请使用更安全的计算方式如ast.literal_eval或专用库这里仅为演示。 result eval(expression) return f计算结果为{result} except Exception as e: return f计算错误{e} # 3. 准备工具列表和提示词模板 tools [calculator] prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的助手可以回答一般问题并且拥有一个计算器工具。如果用户的问题涉及数学计算请务必使用计算器工具来获得精确答案。), (placeholder, {chat_history}), (human, {input}), (placeholder, {agent_scratchpad}), ]) # 4. 创建Agent agent create_tool_calling_agent(llmllm, toolstools, promptprompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) # verboseTrue 会打印思考过程 # 5. 运行Agent if __name__ __main__: # 测试用例1需要计算 result1 agent_executor.invoke({input: 请问 (12 34) * 5.6 等于多少}) print(用户问题请问 (12 34) * 5.6 等于多少) print(fAgent回答{result1[output]}\n) # 测试用例2普通问答 result2 agent_executor.invoke({input: 你是谁}) print(用户问题你是谁) print(fAgent回答{result2[output]}\n) # 测试用例3复杂计算 result3 agent_executor.invoke({input: 计算圆的面积如果半径是7.5厘米。}) print(用户问题计算圆的面积如果半径是7.5厘米。) print(fAgent回答{result3[output]})运行并观察python simple_calculator_agent.py如果一切正常你会看到类似以下的输出其中包含了Agent的“思考过程”因为它设置了verboseTrue Entering new AgentExecutor chain... 用户需要计算一个数学表达式。我应该使用计算器工具。 Action: calculator Action Input: (12 34) * 5.6 Observation: 计算结果为257.6 Thought: 我已经得到了计算结果可以回答用户了。 Final Answer: (12 34) * 5.6 的计算结果是 257.6。 用户问题请问 (12 34) * 5.6 等于多少 Agent回答: (12 34) * 5.6 的计算结果是 257.6。这个简单的例子包含了Agent的核心要素LLM大脑、Prompt指令、Tools工具、Executor执行器。看到这个输出你的基础环境就过关了。2.3 常见环境踩坑点API Key 错误或网络问题 最常见的错误。确保Key正确、有余额并且网络能访问API端点。国内调用国外API可能需要配置网络环境但严禁讨论任何相关工具和方法请使用国内合规的模型服务。依赖版本冲突 LangChain等框架更新很快。如果遇到奇怪报错先检查pip list看关键包langchain,langchain-openai,openai等版本是否兼容。建议使用项目要求的特定版本或锁定较新的稳定版本。本地模型部署失败 如果选择Ollama确保下载的模型名称正确且显存足够。7B模型通常需要至少8GB GPU显存。可以通过ollama run qwen2.5:7b测试模型是否能正常对话。工具定义错误tool装饰的函数必须有清晰的文档字符串docstringAgent依赖它来决定是否调用该工具。输入参数的类型提示如expression: str也很重要。3. 从Demo到项目设计并实现一个“智能数据查询助手”现在我们来做一个更接近实战的小项目。目标构建一个Agent它能理解用户用自然语言提出的数据查询需求自动将其转换为SQL语句查询数据库并将结果以自然语言总结出来。3.1 项目定义与工具准备项目目标用户问“上个月销售额最高的产品是什么”Agent应能理解“上个月”是时间范围“销售额最高”是排序和聚合然后生成SQLSELECT product_name, SUM(sales) FROM orders WHERE order_date ‘2024-04-01’ GROUP BY product_name ORDER BY SUM(sales) DESC LIMIT 1执行查询并返回“上个月销售额最高的产品是XXX总销售额为YYY元。”我们需要三个核心工具数据库连接与查询工具 接收SQL返回查询结果。数据库Schema查看工具 让Agent知道数据库里有哪些表、字段及其含义否则它无法生成正确的SQL。日期处理工具可选但重要 将“上个月”、“本周”等自然语言转换为具体的日期范围。我们先模拟一个简单的SQLite数据库和几张表。# 文件setup_database.py import sqlite3 import pandas as pd from datetime import datetime, timedelta # 创建并连接数据库 conn sqlite3.connect(sales_data.db) cursor conn.cursor() # 创建产品表 cursor.execute( CREATE TABLE IF NOT EXISTS products ( product_id INTEGER PRIMARY KEY, product_name TEXT NOT NULL, category TEXT ) ) # 创建订单表 cursor.execute( CREATE TABLE IF NOT EXISTS orders ( order_id INTEGER PRIMARY KEY, product_id INTEGER, sales REAL, order_date DATE, FOREIGN KEY (product_id) REFERENCES products (product_id) ) ) # 插入模拟数据 products_data [(1, 智能手机X, 电子产品), (2, 蓝牙耳机Y, 电子产品), (3, 咖啡机Z, 家电), (4, 办公椅A, 家具)] cursor.executemany(INSERT OR IGNORE INTO products VALUES (?,?,?), products_data) # 生成最近3个月的模拟订单数据 orders_data [] start_date datetime.now() - timedelta(days90) for i in range(1, 101): product_id (i % 4) 1 sales round(100 (i * 10) % 500, 2) order_date (start_date timedelta(days(i % 90))).strftime(%Y-%m-%d) orders_data.append((i, product_id, sales, order_date)) cursor.executemany(INSERT OR IGNORE INTO orders VALUES (?,?,?,?), orders_data) conn.commit() print(数据库‘sales_data.db’及模拟数据已创建。) print(产品表样例) print(pd.read_sql_query(SELECT * FROM products LIMIT 5, conn)) print(\n订单表样例) print(pd.read_sql_query(SELECT * FROM orders LIMIT 5, conn)) conn.close()运行这个脚本创建你的测试数据库。3.2 构建智能体集成工具与设计提示词现在我们来编写核心的Agent。我们将使用LangChain的create_sql_agent这是一个高级封装它内部已经集成了生成SQL、执行、纠错等逻辑非常适合这个场景。# 文件sql_query_agent.py import os from langchain_openai import ChatOpenAI from langchain_community.utilities import SQLDatabase from langchain_community.agent_toolkits import create_sql_agent from langchain_community.tools import QuerySQLDataBaseTool from datetime import datetime, timedelta # 1. 设置大模型同上 os.environ[OPENAI_API_KEY] your-deepseek-api-key llm ChatOpenAI(modeldeepseek-chat, base_urlhttps://api.deepseek.com/v1, temperature0) # 2. 连接数据库 db SQLDatabase.from_uri(sqlite:///sales_data.db) # 3. 创建SQL Agent agent_executor create_sql_agent( llmllm, dbdb, agent_typeopenai-tools, # 使用工具调用类型的Agent verboseTrue, # 以下是增强Agent能力的额外提示词 prefix你是一个专业的数据库分析师。你的任务是根据用户的问题查询数据库并给出清晰、准确的答案。 请遵循以下规则 1. 首先理解用户问题中的时间范围。如果用户提到“今天”、“昨天”、“本周”、“本月”、“上个月”等你需要将其转换为具体的SQL日期条件。 - 例如‘今天’ 指的是 ‘{today_date}’。 - ‘上个月’ 指的是 从 ‘{last_month_start}’ 到 ‘{last_month_end}’。 我会在系统提示词中为你动态更新这些日期 2. 在生成SQL前先查看相关的数据库表结构schema确保你引用的表名和字段名是正确的。 3. 生成的SQL语句必须语法正确并且只查询必要的数据。 4. 如果查询结果为空或出错请分析原因并尝试调整查询条件或告知用户可能的原因。 5. 最终答案应该用自然语言总结查询结果不要直接输出原始的SQL结果集。 ) # 4. 动态生成包含今日日期的提示词 def get_dynamic_prefix(): today datetime.now().strftime(%Y-%m-%d) last_month_start (datetime.now().replace(day1) - timedelta(days1)).replace(day1).strftime(%Y-%m-%d) last_month_end (datetime.now().replace(day1) - timedelta(days1)).strftime(%Y-%m-%d) prefix f你是一个专业的数据库分析师。你的任务是根据用户的问题查询数据库并给出清晰、准确的答案。 请遵循以下规则 1. 首先理解用户问题中的时间范围。如果用户提到“今天”、“昨天”、“本周”、“本月”、“上个月”等你需要将其转换为具体的SQL日期条件。 - 例如‘今天’ 指的是 ‘{today}’。 - ‘上个月’ 指的是 从 ‘{last_month_start}’ 到 ‘{last_month_end}’。 2. 在生成SQL前先查看相关的数据库表结构schema确保你引用的表名和字段名是正确的。 3. 生成的SQL语句必须语法正确并且只查询必要的数据。 4. 如果查询结果为空或出错请分析原因并尝试调整查询条件或告知用户可能的原因。 5. 最终答案应该用自然语言总结查询结果不要直接输出原始的SQL结果集。 return prefix # 由于create_sql_agent的prefix在创建时固定我们可以通过更新agent_executor.agent.llm_chain.prompt的方式动态修改这里为了简化我们先使用一个静态前缀。 # 在实际更复杂的场景中你可能需要自定义一个更灵活的Agent。 if __name__ __main__: # 测试几个查询 queries [ 总共有多少条订单记录, 上个月销售额最高的产品是什么, 咖啡机Z这个月的总销售额是多少, 哪个产品类别的总销售额最高, ] for query in queries: print(f\n{*50}) print(f用户问题{query}) print(f{*50}) try: # 注意由于我们使用了静态前缀关于“上个月”的日期可能不准确这正是一个可以改进的点。 result agent_executor.invoke({input: query}) print(fAgent回答{result[output]}) except Exception as e: print(f执行出错{e})运行这个Agentpython sql_query_agent.py观察输出。一个成熟的SQL Agent会先查看数据库Schema然后生成SQL执行并返回总结。你可能会看到它生成的SQL类似SELECT p.product_name, SUM(o.sales) as total_sales FROM orders o JOIN products p ON o.product_id p.product_id WHERE o.order_date BETWEEN ‘2024-04-01’ AND ‘2024-04-30’ GROUP BY p.product_name ORDER BY total_sales DESC LIMIT 13.3 项目深化错误处理与能力扩展一个基础版本跑通后就要考虑企业级需求了SQL注入与安全性 我们的Agent直接执行模型生成的SQL。这在生产环境是极度危险的必须对生成的SQL进行严格的校验和限制例如只允许SELECT禁止DROP、DELETE等。可以使用sqlparse库进行解析或使用数据库用户的只读权限。复杂日期/语义处理 我们的简单前缀提示可能不够。更好的做法是专门设计一个“时间解析工具”。这个工具接收“上个月”、“本周三”等字符串返回标准的日期范围或SQL WHERE条件片段然后让Agent在生成SQL时调用这个工具的结果。查询优化与解释 对于复杂的查询Agent可以尝试解释它为什么这么写或者当查询很慢时建议添加索引。结果可视化 可以集成一个工具当用户问“画一个趋势图”时调用matplotlib或plotly生成图表并保存为文件然后将文件路径或图片描述返回给用户。这个“智能数据查询助手”项目麻雀虽小五脏俱全。它涉及了自然语言理解、工具调用SQL执行、状态管理日期上下文、错误处理SQL错误。把它做扎实你对Agent开发的理解会上一个大台阶。4. 进阶实战构建多智能体工作流处理复杂任务单智能体处理线性任务够了但面对“调研一个主题并撰写报告”这类需要检索、分析、写作、审核多步骤的任务就需要工作流了。我们使用LangGraph来构建一个简单的“研究助手”工作流。场景用户输入一个研究主题如“大模型智能体的最新发展趋势”工作流自动执行1. 网络搜索 - 2. 内容总结 - 3. 报告撰写 - 4. 报告评分。如果评分过低则重新搜索或总结。4.1 定义工作流中的节点智能体我们将定义四个节点每个节点可以看作一个具备特定能力的智能体或一个函数。# 文件research_workflow.py import os from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI from langchain_community.tools import TavilySearchResults from langchain_core.messages import HumanMessage, SystemMessage import operator # 0. 设置环境和模型 os.environ[OPENAI_API_KEY] your-openai-or-other-api-key os.environ[TAVILY_API_KEY] your-tavily-api-key # 用于搜索可在Tavily官网申请免费额度 llm ChatOpenAI(modelgpt-4o-mini, temperature0.7) # 使用一个适合创意和总结的模型 # 1. 定义工作流的状态State class ResearchState(TypedDict): # 消息历史 messages: Annotated[List, add_messages] # 研究主题 topic: str # 搜索到的原始内容 raw_search_results: str # 总结后的内容 summarized_content: str # 生成的报告 report: str # 报告评分 (1-10) report_score: int # 2. 定义各个节点函数 def search_node(state: ResearchState) - ResearchState: 节点1搜索。使用Tavily搜索工具获取最新信息。 print(f[Search Node] 正在搜索主题: {state[topic]}) search_tool TavilySearchResults(max_results3) # 限制3条结果 search_query f{state[topic]} latest developments 2024 try: results search_tool.invoke(search_query) # 将结果拼接成字符串 combined_results \n---\n.join([f来源{r.get(title, N/A)}\n内容{r.get(content, N/A)} for r in results]) return {raw_search_results: combined_results} except Exception as e: print(f搜索失败: {e}) return {raw_search_results: f搜索时出错: {e}} def summarize_node(state: ResearchState) - ResearchState: 节点2总结。基于搜索到的原始内容生成一份简洁的摘要。 print(f[Summarize Node] 正在总结搜索内容...) if not state[raw_search_results] or 出错 in state[raw_search_results]: return {summarized_content: 无有效搜索内容可供总结。} prompt f 你是一个专业的研究助理。请根据以下搜索到的原始材料提炼出关于“{state[topic]}”的核心观点、关键事实和主要趋势。 摘要要求简洁、客观、信息密度高字数在300字以内。 原始材料 {state[raw_search_results]} 摘要 messages [SystemMessage(content你是一个专业的总结者。), HumanMessage(contentprompt)] response llm.invoke(messages) return {summarized_content: response.content} def write_report_node(state: ResearchState) - ResearchState: 节点3撰写报告。基于总结的内容撰写一份结构完整的简短报告。 print(f[Write Report Node] 正在撰写报告...) if not state[summarized_content]: return {report: 无法撰写报告因为缺少总结内容。} prompt f 你是一名技术作家。请基于以下关于“{state[topic]}”的摘要撰写一份结构清晰、内容充实的简短报告。 报告应包含引言、主要发展趋势、当前面临的挑战、未来展望等部分。总字数控制在500字左右。 摘要内容 {state[summarized_content]} 报告 messages [SystemMessage(content你是一名优秀的报告撰写者。), HumanMessage(contentprompt)] response llm.invoke(messages) return {report: response.content} def evaluate_report_node(state: ResearchState) - ResearchState: 节点4评估报告。对生成的报告进行质量评分并决定工作流是否结束。 print(f[Evaluate Node] 正在评估报告质量...) if not state[report]: return {report_score: 0} prompt f 你是一名严格的评审。请对以下关于“{state[topic]}”的技术报告进行评分1-10分10分为最高。 评分标准 - 结构清晰度 (2分) - 信息准确性与相关性 (3分) - 内容深度与洞察力 (3分) - 语言流畅度 (2分) 请只输出一个整数分数。 报告内容 {state[report][:1000]} # 限制长度以防token超限 分数 messages [SystemMessage(content你是一名公正的评审。), HumanMessage(contentprompt)] response llm.invoke(messages) try: score int(response.content.strip()) except: score 5 # 解析失败则给中间分 print(f 评估得分: {score}/10) # 将分数存入状态 return {report_score: score} def should_continue(state: ResearchState) - str: 条件判断函数根据评分决定下一步是重新搜索还是结束。 score state.get(report_score, 0) # 如果评分低于6分我们让工作流跳回搜索节点重新开始否则结束。 # 注意这是一个简单演示实际中可能设置最大循环次数避免死循环。 if score 6: print(f 报告评分{score}分过低将重新开始研究流程。) return search # 返回下一个要执行的节点名 else: print(f 报告评分{score}分合格工作流结束。) return END # 3. 构建工作流图 workflow StateGraph(ResearchState) # 添加节点 workflow.add_node(search, search_node) workflow.add_node(summarize, summarize_node) workflow.add_node(write_report, write_report_node) workflow.add_node(evaluate, evaluate_report_node) # 设置边的起点 workflow.set_entry_point(search) # 添加条件边 workflow.add_conditional_edges( evaluate, # 从评估节点出发 should_continue, # 根据这个函数的返回值决定去哪 { search: search, # 如果返回search则跳回搜索节点 END: END # 如果返回END则结束 } ) # 添加普通边线性执行部分 workflow.add_edge(search, summarize) workflow.add_edge(summarize, write_report) workflow.add_edge(write_report, evaluate) # 编译图 app workflow.compile() # 4. 运行工作流 if __name__ __main__: # 初始化状态 initial_state: ResearchState { messages: [], topic: 大模型智能体AI Agent的最新发展趋势, raw_search_results: , summarized_content: , report: , report_score: 0 } print(开始执行研究助手工作流...) print(f研究主题: {initial_state[topic]}) print(- * 50) # 执行图 final_state app.invoke(initial_state, config{recursion_limit: 3}) # 限制递归深度为3次防止无限循环 print(\n *50) print(工作流执行完毕) print(f最终报告评分: {final_state.get(report_score)}/10) print(\n生成的报告内容预览) print(final_state.get(report, )[:500] ...) # 打印前500字符4.2 理解工作流的关键设计这个例子虽然简单但体现了多智能体工作流的核心思想状态State共享 我们定义了一个ResearchState字典所有节点都读写这个状态。这是智能体之间通信的主要方式。节点Node专精 每个节点只做一件事搜索、总结、写作、评估符合单一职责原则易于调试和替换。条件路由Conditional Edgeshould_continue函数根据报告评分动态决定工作流是结束还是重新开始。这赋予了工作流“循环”和“自我修正”的能力。图Graph结构 工作流被明确定义为一张图节点是处理单元边是执行顺序或条件逻辑。这比写一堆if-else嵌套要清晰和强大得多。要运行这个工作流你需要一个可用的OpenAI格式的API示例中用了gpt-4o-mini你可以换成其他模型。一个 Tavily 的API Key用于搜索它有免费额度。你也可以替换成其他搜索工具如Serper API、Google Search API等。4.3 企业级工作流的考量在实际项目中你还需要考虑错误处理与重试 任何一个节点如搜索API调用失败都可能出错。需要在节点内部或图层面添加重试逻辑和错误处理边。异步与并行 如果“搜索”和“查阅内部文档”可以同时进行就应该设计成并行节点以提高效率。LangGraph支持异步和并行执行。人工审核节点 在关键节点如报告发布前插入一个“人工审核”节点等待人工确认后再继续。持久化与可观测性 将每次运行的状态、中间结果和最终输出保存到数据库方便追溯和调试。使用langgraph的检查点Checkpoint功能可以实现状态持久化和断点续跑。5. 项目复盘与面试准备从会做到会讲完成几个项目后如何把它们变成你的竞争力面试官不会只看你的代码他更想听你为什么这么设计遇到了什么问题以及如何解决的。5.1 为你的项目准备“叙事线”针对任何一个你完成的Agent项目准备好回答以下问题项目背景与目标 一句话说清楚这个Agent是干什么的解决了什么痛点。例如“我开发了一个智能数据查询助手让业务人员能用自然语言直接查询数据库无需学习SQL。”架构与核心组件大脑LLM 你选用哪个模型为什么考虑成本、性能、中文能力、API稳定性。工具Tools 你为Agent配备了哪些工具为什么选这些如何保证工具调用的安全性和稳定性例如SQL注入防护、API限流与重试。记忆Memory 你用了哪种记忆会话记忆ConversationBufferMemory还是长期记忆向量数据库如何管理记忆的token消耗编排Orchestration 是单智能体还是工作流如果是工作流图是怎么设计的节点间如何传递数据关键挑战与解决方案提示工程Prompt Engineering 如何设计System Prompt和Few-shot Examples来让Agent更听话、更准确工具调用可靠性 如何处理工具调用失败网络超时、API错误有没有实现fallback机制成本与延迟优化 如何减少不必要的LLM调用或Token消耗例如缓存、思维链压缩、更精准的工具选择。评估与迭代 如何评估你的Agent效果有没有建立测试集如何根据反馈迭代Prompt和工具部署与监控如何部署你的AgentFastAPI封装成服务Streamlit做界面如何监控它的运行状态日志记录、关键指标如调用次数、成功率、平均响应时间。5.2 针对Agent开发的常见面试题除了项目本身你还需要了解一些通用知识Agent vs Chain vs Pipeline 说说你对它们的理解。核心区别在于Agent有决策能力决定下一步做什么调用哪个工具而Chain是固定的序列Pipeline可能更偏向数据流。ReAct vs Plan-and-Execute 两种主流的Agent推理框架。ReActReasoning Acting是边想边做每一步思考后决定行动Plan-and-Execute是先制定完整计划再执行。前者灵活后者可能更高效。你项目中用的是哪种思路LangChain, LlamaIndex, AutoGen等框架的优劣 不必全部精通但要对最常用的如LangChain有深入理解并了解其他框架的侧重点如LlamaIndex擅长检索AutoGen擅长多智能体对话。大模型幻觉Hallucination与缓解 Agent严重依赖LLM如何应对LLM胡说八道常用方法有要求提供引用来源检索增强生成RAG、设置验证步骤让另一个Agent或规则检查输出、在Prompt中强调“不知道就说不知道”。安全性考量 除了前面提到的SQL注入还要考虑工具权限控制不能让Agent随意调用删除文件的工具、用户输入过滤、输出内容安全审查、API密钥管理等。5.3 学习路线与资源建议回到标题中的“100个项目”其价值在于提供广泛的场景。但你的学习不应是线性的。更高效的方式是掌握核心范式 通过2-3个不同类型的项目如带工具调用的单智能体、基于RAG的问答Agent、多智能体工作流彻底吃透Agent的核心概念和主流框架LangChain/LangGraph的使用。垂直深入 选择一个你感兴趣的垂直领域如智能客服、数据分析、自动化办公、游戏NPC用Agent技术去解决该领域的一个具体问题并做深做透。关注底层与前沿 了解一些底层框架如DSPy、MCP和前沿架构如CrewAI、Microsoft Autogen理解其设计思想。关注开源模型如DeepSeek-V2、Qwen2.5的Agent能力进展。工程化实践 学习如何将你的Agent项目容器化Docker、编写测试用例、配置CI/CD、设计监控告警。这是从“玩具”到“产品”的关键一步。记住“100个项目”的目标不是刷数量而是通过足够的练习让你在面对一个全新的、模糊的需求时能快速将其分解为Agent可解决的子问题并选择合适的技术组件进行搭建。这才是企业真正需要的能力。