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 系统设计与模块划分

这个助手需要以下模块:

  1. 大纲生成智能体:调用LLM,根据主题生成研究大纲(引言、相关工作、方法、实验、结论)。
  2. 文献查询工具:模拟从学术数据库(如arXiv)查询相关论文标题。由于真实API需要密钥且网络可能受限,我们这里用一个本地JSON文件模拟一个论文库。
  3. 报告整合智能体:将大纲和查询到的文献列表整合成一份完整的建议报告。

我们使用 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智能体开发目前还是一个工程实践性极强的领域,动手做,遇到问题解决问题,是最好的学习方式。当你看到自己打造的智能体流畅地理解、规划、执行任务时,那种成就感是无与伦比的。

Logo

北京人形旗下天工造物具身智能开源社区,聚焦具身天工与慧思开物两大平台

更多推荐