【实战指南】从零到一:手把手教你打造AI复合智能体
1. 环境准备:从零开始的AI智能体开发环境
要打造一个能跑起来的AI复合智能体,第一步就是把“厨房”收拾好。这里说的厨房,就是你的开发环境。我见过很多新手朋友,一上来就急着写代码、调模型,结果被各种依赖报错、版本冲突搞得焦头烂额,热情一下就浇灭了。所以,咱们先别急,花点时间把地基打牢。
首先,你需要一个趁手的“工作台”。我强烈推荐使用 Python 3.10 或 3.11 版本,这是目前大多数AI框架最稳定的支持版本。太老的版本可能缺少一些新特性,太新的版本(比如3.12)有时会遇到一些库的兼容性问题。安装好Python后,第一件事就是创建一个独立的虚拟环境。这就像给你的项目单独准备一个工具箱,里面的工具(各种Python包)不会和其他项目混在一起,避免“打架”。用 venv 或者 conda 都可以,我个人习惯用 venv,因为它轻量且是Python自带的。
# 创建项目文件夹并进入
mkdir my_ai_agent && cd my_ai_agent
# 创建虚拟环境
python -m venv venv
# 激活虚拟环境(Windows用 venv\Scripts\activate)
source venv/bin/activate
激活后,你的命令行前面会出现 (venv) 的提示,这就说明你已经在这个独立的环境里了。接下来,安装最核心的“脚手架”——LangChain。LangChain是构建AI智能体的瑞士军刀,它把调用大模型、管理工具、处理记忆这些复杂的事情都封装好了,让我们能更专注于业务逻辑。我们同时也会安装一些必要的辅助库。
pip install langchain langchain-community langchain-core
光有框架还不够,我们需要一个“大脑”,也就是大语言模型。对于初学者,我建议从调用云端API开始,比如DeepSeek、智谱AI或者百度的千帆,它们提供了稳定、易用的接口,省去了自己部署模型的巨大开销。这里以DeepSeek为例,你需要先去其官网注册获取一个API Key。然后安装对应的SDK。
pip install langchain-deepseek
最后,我们还需要一个“记事本”,用来管理智能体的记忆。简单的对话记忆可以用LangChain自带的,但如果想实现更复杂的、基于向量检索的记忆(也就是RAG),就需要一个向量数据库。入门阶段,我们可以先用Chroma,它轻量且易于集成。
pip install chromadb
好了,敲完这几行命令,你的基础开发环境就准备好了。这个过程可能会因为网络问题需要一点耐心,但每一步都是必经之路。我建议你每成功安装一个包,都简单测试一下,比如 python -c “import langchain; print(‘ok’)”,确保没有导入错误。环境搭建是实战的第一步,也是排除万难的第一步,走稳了,后面的路才会顺。
2. 核心模块拆解:打造智能体的“五官”与“四肢”
环境就绪,现在我们来聊聊智能体到底由哪些部分组成。你可以把它想象成一个有“五官”和“四肢”的智能机器人。感知模块就是它的耳朵和眼睛,负责理解用户的输入。决策模块是它的大脑,负责思考和分析。行动模块是它的手和脚,负责执行具体的任务。记忆模块则是它的笔记本,用来记住之前的对话和状态。我们一个个来搭建。
2.1 感知与决策:连接大模型,让智能体“听懂人话”
感知和决策的核心,就是与大语言模型(LLM)的交互。在LangChain里,这非常简单。我们首先需要初始化一个LLM对象,它就是智能体思考的引擎。
from langchain_deepseek import ChatDeepSeek
from langchain_core.prompts import ChatPromptTemplate
# 替换成你的真实API Key,建议通过环境变量管理,不要硬编码在代码里
import os
os.environ[“DEEPSEEK_API_KEY”] = “your-api-key-here”
# 初始化DeepSeek模型
llm = ChatDeepSeek(model=“deepseek-chat”, temperature=0.1)
# temperature控制创造性,0.1表示更确定、更保守的回答,适合任务执行
# 创建一个简单的提示词模板
prompt_template = ChatPromptTemplate.from_messages([
(“system”, “你是一个乐于助人的AI助手。”),
(“human”, “{user_input}”)
])
# 将提示词和模型组合成一条链
chain = prompt_template | llm
# 测试一下
response = chain.invoke({“user_input”: “你好,介绍一下你自己。”})
print(response.content)
运行这段代码,你应该能看到模型返回的自我介绍。这里的 | 符号是LangChain LCEL语法,代表“管道”,意思是将前一个组件的输出作为后一个组件的输入,非常直观。temperature 参数很重要,在构建执行任务的智能体时,我们通常把它设得低一点(比如0.1-0.3),这样模型的输出更稳定、更可预测;如果是创作类任务,可以调高来增加多样性。
注意:API Key是敏感信息,千万不要上传到GitHub等公开仓库。最佳实践是使用
python-dotenv库从.env文件读取,或者直接设置系统环境变量。
2.2 行动与工具:给智能体装上“手和脚”
智能体光会思考可不行,它得能“做事”。这就是工具(Tools)的作用。工具可以是任何东西:查询天气、计算数学、搜索网络、操作数据库等等。我们来创建一个最简单的工具——计算器。
from langchain.tools import tool
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
# 使用 @tool 装饰器定义一个工具
@tool
def calculate(expression: str) -> str:
“”“计算一个数学表达式的结果。例如:’calculate(2+3*4)’“”“
try:
# 警告:实际生产中,直接eval有安全风险,这里仅作演示。
# 更安全的做法是使用 ast.literal_eval 或专门的数学解析库。
result = eval(expression)
return f“表达式 {expression} 的计算结果是:{result}”
except Exception as e:
return f“计算出错:{e}”
# 准备工具列表
tools = [calculate]
# 为智能体构建一个更详细的提示词,告诉它可以使用工具
agent_prompt = ChatPromptTemplate.from_messages([
(“system”, “””
你是一个专业的数学助手。你可以使用工具来帮助用户计算数学表达式。
如果你需要计算,请务必调用工具。
用户问题:{input}
“””),
(“placeholder”, “{agent_scratchpad}”) # 这是一个占位符,用于记录智能体思考过程
])
# 创建智能体
agent = create_tool_calling_agent(llm=llm, tools=tools, prompt=agent_prompt)
# 创建智能体执行器
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
# 测试带工具的智能体
result = agent_executor.invoke({“input”: “请问125乘以88等于多少?”})
print(result[“output”])
当你运行这段代码,并设置 verbose=True 时,你会在控制台看到详细的思考过程:智能体先决定要调用 calculate 工具,然后生成正确的调用参数 “125*88”,工具执行后返回结果,最后智能体将这个结果组织成自然语言回复给用户。这个过程就是经典的 ReAct(Reasoning + Acting) 模式。现在,你的智能体就有了一只可以执行计算任务的“手”。你可以用同样的方式,定义更多工具,比如调用一个真实的天气API,或者查询数据库。
2.3 记忆系统:让对话有上下文,不再“金鱼脑”
没有记忆的智能体,每次对话都是全新的开始,用户会说“你刚才说的那个事情”,而它一脸茫然。所以我们需要给智能体加上记忆。LangChain 提供了多种记忆类型,最简单常用的是 ConversationBufferMemory,它会像缓存一样记住最近的对话历史。
from langchain.memory import ConversationBufferMemory
# 初始化记忆
memory = ConversationBufferMemory(memory_key=“chat_history”, return_messages=True)
# 我们需要更新提示词,把记忆放进去
agent_prompt_with_memory = ChatPromptTemplate.from_messages([
(“system”, “你是一个友好的聊天助手。请根据对话历史来回答。”),
MessagesPlaceholder(variable_name=“chat_history”), # 历史对话占位符
(“human”, “{input}”),
MessagesPlaceholder(variable_name=“agent_scratchpad”)
])
# 重新创建智能体和执行器,这次绑定记忆
agent_with_memory = create_tool_calling_agent(llm=llm, tools=tools, prompt=agent_prompt_with_memory)
agent_executor_with_memory = AgentExecutor(
agent=agent_with_memory,
tools=tools,
memory=memory,
verbose=True
)
# 进行多轮对话
print(“第一轮:”)
result1 = agent_executor_with_memory.invoke({“input”: “我叫小明。”})
print(result1[“output”])
print(“\n第二轮:”)
result2 = agent_executor_with_memory.invoke({“input”: “你还记得我叫什么名字吗?”})
print(result2[“output”])
你会发现,在第二轮对话中,智能体能够正确回答出你的名字,因为它从 chat_history 中读取了上一轮的信息。这就是短期记忆。对于更复杂的、需要从大量文档中查找信息的记忆,就需要用到我们之前安装的向量数据库和RAG技术了,那相当于给智能体配了一个外部知识库,我们会在后面进阶部分详细展开。
3. 从单兵到军团:构建多智能体协作系统
单个智能体能力再强,也有极限。面对复杂任务,比如“帮我规划一个旅行行程,包括订机票、查天气、推荐景点”,最好的办法是组建一个“特种部队”,让不同的智能体各司其职,协同工作。这就是多智能体系统。听起来很高大上,但用 LangGraph 这个框架,我们可以像画流程图一样把它构建出来。
3.1 认识LangGraph:智能体协作的“交通指挥中心”
LangGraph 是 LangChain 生态中专门用于构建有状态、多环节工作流的库。它把每个智能体或操作看作一个“节点”(Node),节点之间通过“边”(Edge)连接,数据在图中按规则流动。我们来实现一个简单的双智能体协作场景:一个“策划者”负责分解任务,一个“执行者”负责调用工具。
首先,确保安装 LangGraph:pip install langgraph。然后,我们定义两个简单的智能体角色。
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
# 首先定义整个图的状态结构,这是一个共享的白板
class AgentState(TypedDict):
# 用户原始问题
question: str
# 策划者生成的计划
plan: str
# 执行者生成的结果
answer: str
# 记录已经走了多少步,防止死循环
steps: Annotated[int, operator.add]
# 定义“策划者”节点函数
def planner_node(state: AgentState):
print(f“【策划者】正在分析问题:{state[‘question’]}”)
# 这里为了简化,我们写死逻辑。实际应用中,这里应该调用一个LLM来生成计划。
if “天气” in state[“question”]:
plan = “这是一个查询天气的问题,需要调用天气工具。”
elif “计算” in state[“question”]:
plan = “这是一个计算问题,需要调用计算器工具。”
else:
plan = “这是一个普通问题,我直接回答。”
return {“plan”: plan, “steps”: 1}
# 定义“执行者”节点函数
def executor_node(state: AgentState):
print(f“【执行者】收到计划:{state[‘plan’]}”)
# 根据计划执行不同操作
if “天气工具” in state[“plan”]:
# 模拟调用天气API
answer = “执行者调用天气API:北京今天晴,15-25度。”
elif “计算器工具” in state[“plan”]:
# 模拟调用计算器
answer = “执行者调用计算器:结果是11000。”
else:
answer = f“执行者直接回复:关于‘{state[‘question’]}’,这是一个好问题。”
return {“answer”: answer, “steps”: 1}
# 开始构建图
workflow = StateGraph(AgentState)
# 添加节点
workflow.add_node(“planner”, planner_node)
workflow.add_node(“executor”, executor_node)
# 设置入口点
workflow.set_entry_point(“planner”)
# 定义边:策划者完成后,总是交给执行者
workflow.add_edge(“planner”, “executor”)
# 执行者完成后,结束流程
workflow.add_edge(“executor”, END)
# 编译图
app = workflow.compile()
现在,我们的协作流程图就建好了。运行它看看效果:
# 运行图
initial_state = {“question”: “北京今天的天气怎么样?”, “steps”: 0}
final_state = app.invoke(initial_state)
print(f“\n最终答案:{final_state[‘answer’]}”)
print(f“总步骤数:{final_state[‘steps’]}”)
在控制台,你会看到清晰的节点执行日志。策划者先分析问题类型,然后执行者根据计划去“执行”。虽然这里的逻辑是写死的,但你已经掌握了核心范式:定义状态、创建节点、连接节点、编译运行。在实际项目中,你可以把 planner_node 和 executor_node 内部替换成真正的、带有LLM和工具调用的完整智能体,图的边也可以根据条件动态决定下一步走哪里,从而实现异常复杂的业务流程。
3.2 引入通信协议:让智能体高效“对话”
当智能体多起来,它们之间如何通信就成了问题。你不能让它们胡乱喊话。这时就需要一个简单的通信协议。我们可以基于状态(State)来模拟。比如,每个智能体把自己的“发言”追加到一个共享的 messages 列表里,其他智能体可以读取。
class MultiAgentState(TypedDict):
task: str
messages: Annotated[list, operator.add] # 这是一个特殊的注解,表示列表是追加的
final_result: str
def agent_a_node(state: MultiAgentState):
new_message = f“AgentA收到任务:{state[‘task’]}。我认为第一步是市场调研。”
return {“messages”: [new_message]}
def agent_b_node(state: MultiAgentState):
# 读取上一条消息
last_msg = state[“messages”][-1] if state[“messages”] else “”
new_message = f“AgentB看到‘{last_msg}’。我同意,并补充需要分析竞品。”
return {“messages”: [new_message]}
def summarizer_node(state: MultiAgentState):
# 汇总所有消息,形成最终结论
summary = “ | “.join(state[“messages”])
return {“final_result”: f“讨论总结:{summary}”}
在这个模型里,messages 列表就是一个简单的通信通道。更工程化的实现会使用像 A2A(Agent-to-Agent) 这样的协议,它定义了标准的消息格式(发送者、接收者、消息类型、内容体),确保智能体间的交互清晰、可追溯、可调试。对于入门项目,从共享状态开始理解概念就足够了。
4. 项目实战:组装一个个人科研助手智能体
理论讲得再多,不如动手做一个。我们来整合前面学到的所有知识,构建一个稍微复杂点的 个人科研助手智能体。它的功能是:当你输入一个研究主题(比如“大语言模型在医疗诊断中的应用”),它能帮你生成一个初步的研究大纲,并模拟查询相关文献。
4.1 系统设计与模块划分
这个助手需要以下模块:
- 大纲生成智能体:调用LLM,根据主题生成研究大纲(引言、相关工作、方法、实验、结论)。
- 文献查询工具:模拟从学术数据库(如arXiv)查询相关论文标题。由于真实API需要密钥且网络可能受限,我们这里用一个本地JSON文件模拟一个论文库。
- 报告整合智能体:将大纲和查询到的文献列表整合成一份完整的建议报告。
我们使用 LangGraph 来编排这个流程。
4.2 逐步实现
首先,创建一个模拟的论文数据库 papers.json:
[
{“title”: “LLM for Medical Report Generation”, “topic”: “medical, nlp”},
{“title”: “A Survey of AI in Healthcare”, “topic”: “medical, survey”},
{“title”: “Diagnostic Accuracy of Transformer Models”, “topic”: “medical, diagnosis”},
{“title”: “Vision-Language Models in Radiology”, “topic”: “medical, vision”}
]
然后,开始编写主程序:
import json
from langchain_deepseek import ChatDeepSeek
from langgraph.graph import StateGraph, END
from typing import TypedDict, List
import operator
# 定义状态
class ResearchState(TypedDict):
research_topic: str
outline: str
found_papers: List[str]
final_report: str
# 初始化LLM
llm = ChatDeepSeek(model=“deepseek-chat”, temperature=0.3)
# 节点1:生成大纲
def outline_generator_node(state: ResearchState):
print(f“【大纲生成器】正在为‘{state[‘research_topic’]}’生成大纲...”)
prompt = f”””
你是一个学术专家。请为以下研究主题生成一个详细的研究大纲,包括:
1. 引言与研究背景
2. 相关工作综述
3. proposed 方法论
4. 预期的实验设计
5. 结论与未来工作
研究主题:{state[‘research_topic’]}
“””
response = llm.invoke(prompt)
return {“outline”: response.content}
# 节点2:查询文献(模拟)
def paper_search_node(state: ResearchState):
print(f“【文献查询】正在查询与‘{state[‘research_topic’]}’相关的文献...”)
# 加载模拟数据库
with open(“papers.json”, “r”) as f:
all_papers = json.load(f)
# 简单关键词匹配(实际应用应用embedding语义搜索)
topic_lower = state[‘research_topic’].lower()
found = []
for paper in all_papers:
if “medical” in topic_lower and “medical” in paper[“topic”]:
found.append(paper[“title”])
# 模拟返回前3篇
return {“found_papers”: found[:3]}
# 节点3:整合报告
def report_integrator_node(state: ResearchState):
print(“【报告整合器】正在生成最终建议报告...”)
papers_str = “\n”.join([f”- {p}” for p in state[‘found_papers’]])
prompt = f”””
请整合以下信息,形成一份给研究人员的完整建议报告:
研究主题:{state[‘research_topic’]}
生成的研究大纲:
{state[‘outline’]}
查询到的相关文献(供参考):
{papers_str}
请以清晰、专业的格式组织报告,先总结大纲要点,然后列出参考文献建议。
“””
response = llm.invoke(prompt)
return {“final_report”: response.content}
# 构建图
workflow = StateGraph(ResearchState)
workflow.add_node(“generate_outline”, outline_generator_node)
workflow.add_node(“search_papers”, paper_search_node)
workflow.add_node(“integrate_report”, report_integrator_node)
# 设置流程:先大纲,再并行查文献,最后整合
workflow.set_entry_point(“generate_outline”)
workflow.add_edge(“generate_outline”, “search_papers”)
workflow.add_edge(“search_papers”, “integrate_report”)
workflow.add_edge(“integrate_report”, END)
app = workflow.compile()
# 运行!
initial_state = {“research_topic”: “大语言模型在医疗诊断中的应用”}
result = app.invoke(initial_state)
print(“\n” + “=”*50)
print(“【最终科研助手报告】”)
print(“=”*50)
print(result[“final_report”])
运行这个程序,你会看到一个完整的端到端流程:生成大纲、查询文献、整合报告。虽然文献查询是模拟的,但整个架构是真实可扩展的。你可以轻松地将 paper_search_node 替换成真正的 arXiv API 调用,或者接入一个向量数据库实现真正的语义检索(RAG)。这就是复合智能体的魅力——通过组装和编排,让多个 specialized 的模块协同完成一项复杂任务。
踩过几次坑之后,我最大的体会是:构建智能体时,不要一开始就追求大而全。从一个能跑通的最小可行产品(MVP)开始,比如先做一个能调用计算器的单一智能体,然后加上记忆,再尝试拆分成两个协作的智能体。每一步都充分测试,理解状态是如何流动的。遇到问题,多利用 verbose=True 查看中间过程,或者把关键状态打印出来。AI智能体开发目前还是一个工程实践性极强的领域,动手做,遇到问题解决问题,是最好的学习方式。当你看到自己打造的智能体流畅地理解、规划、执行任务时,那种成就感是无与伦比的。
更多推荐
所有评论(0)