LangChain LangGraph 教程:agent智能体入门-通过DeepSeek调用乘法函数

介绍

智能体核心公式:

Agent = LLM(DeepSeek 大脑) + Tools(可调用能力) + Prompt/记忆(行为逻辑) + 执行循环(ReAct/LangGraph)
  • DeepSeek:提供推理、思考、工具调用决策(支持 chat / reasoner 两种模式)
  • LangChain:标准化工具、记忆、提示词、Agent 运行器
  • Agent 运行方式:
    • 基础工具调用 Agent:单轮/多轮调用工具,适合简单任务
    • ReAct Agent:思考→行动→观察→再思考,适合多步推理
    • LangGraph Agent:状态机编排,适合复杂、长流程、需要记忆/规划的任务

应用场景

  • Agentic RAG:文档问答 + 联网搜索 + 工具调用
  • 数据处理 Agent:自动读取 CSV → 分析 → 可视化 → 报告
  • 个人助理:日程管理、邮件处理、天气查询、计算器
  • 代码助手:写代码、查文档、运行脚本(需沙箱)

环境准备

1. 安装依赖

python -m venv ds-agent-env
source ds-agent-env/bin/activate  # Windows: ds-agent-env\Scripts\activate

pip install -U langchain langchain-openai langchain-community
pip install python-dotenv requests
# 可选:搜索工具、向量库、LangGraph
pip install duckduckgo-search chromadb langgraph

获取 DeepSeek API Key

  • 官网:https://www.deepseek.com
  • 注册后创建 API Key,充值即可使用
  • 支持模型:
    • deepseek-chat:通用对话,最适合 Agent
    • deepseek-reasoner:强推理,适合复杂数学/逻辑任务
    • deepseek-v4-flash:高速低延迟

环境变量 .env

DEEPSEEK_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
DEEPSEEK_API_BASE=https://api.deepseek.com/v1
DEEPSEEK_MODEL=deepseek-chat

接入 DeepSeek 的两种方式

方式 A:用 ChatOpenAI 兼容(最简单,推荐入门)

DeepSeek 完全兼容 OpenAI 接口格式,可直接用 langchain-openai:

# llm.py
from dotenv import load_dotenv
import os
from langchain_openai import ChatOpenAI

load_dotenv()

llm = ChatOpenAI(
    model=os.getenv("DEEPSEEK_MODEL", "deepseek-chat"),
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url=os.getenv("DEEPSEEK_API_BASE"),
    temperature=0.1,  # 越低越稳定,Agent 建议 0~0.3
    max_tokens=4096
)

# 测试
if __name__ == "__main__":
    res = llm.invoke("你好,你是 DeepSeek 吗?")
    print(res.content)

方式 B:用官方 ChatDeepSeek(保留 reasoner 推理字段)

pip install langchain-deepseek
from langchain_deepseek import ChatDeepSeek

llm = ChatDeepSeek(
    model="deepseek-reasoner",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    api_base="https://api.deepseek.com/v1",
    temperature=0.0
)

编写自定义工具(Agent 的“手脚”)

工具规范

  • 用 @tool 装饰器
  • 函数文档字符串必须清晰(LLM 靠它理解参数)
  • 参数尽量简单(str/int/float)
  • 返回值结构化,方便 LLM 总结

示例:实现计算器工具(ChatOpenAI 兼容)

from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langgraph.prebuilt import create_react_agent

# 1. 自定义乘法工具函数
@tool
def multiply(a: float, b: float) -> float:
    """
    计算两个数字的乘积
    :param a: 第一个乘数
    :param b: 第二个乘数
    """
    return a * b

# 2. 初始化大模型
llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    api_key="你的密钥",
    base_url="接口地址"
)

# 3. 创建智能体,绑定乘法工具
agent = create_react_agent(llm, tools=[multiply])

# 4. 智能体交互调用
if __name__ == "__main__":
    # 自然语言提问,自动调用乘法函数
    res1 = agent.invoke({"messages": [("user", "计算8乘以9等于多少")]})
    print("结果1:", res1["messages"][-1].content)

    res2 = agent.invoke({"messages": [("user", "3.5乘6.2是多少")]})
    print("结果2:", res2["messages"][-1].content)

示例:天气查询 + 计算器工具

# tools.py
from langchain.tools import tool
import requests

# 1. 天气工具
@tool
def get_weather(city: str) -> str:
    """查询指定城市的当前天气。
    参数:
        city: 城市名,如"北京"、"上海"
    """
    # 模拟调用,实际替换为真实天气 API
    mock_data = {
        "北京": "晴,25℃,微风",
        "上海": "多云,28℃,东南风"
    }
    return mock_data.get(city, "未知城市")

# 2. 计算器工具
@tool
def calculator(expression: str) -> str:
    """执行数学计算,支持加减乘除、括号。
    参数:
        expression: 数学表达式字符串,如"256*128"
    """
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return f"计算结果:{result}"
    except Exception as e:
        return f"计算错误:{e}"

tools = [get_weather, calculator]

创建基础工具调用智能体(Tool Calling Agent)

适合:简单问答、单/多工具调用、无需复杂思考

# agent_basic.py
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
from llm import llm
from tools import tools

# 1. 提示词模板(控制 Agent 行为)
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个专业助手,会准确调用工具。回答用中文,简洁清晰。"),
    ("user", "{input}"),
    ("agent_scratchpad", "{agent_scratchpad}")
])

# 2. 创建 Agent
agent = create_tool_calling_agent(llm, tools, prompt)

# 3. 执行器(处理循环调用、错误)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 4. 测试
if __name__ == "__main__":
    result = agent_executor.invoke({"input": "北京天气怎么样?顺便算一下 123*456"})
    print("最终回答:", result["output"])

运行效果:

  • LLM 先决定调用 get_weather("北京")
  • 拿到结果后再调用 calculator("123*456")
  • 最后整合结果输出中文回答

创建 ReAct 智能体(思考→行动→观察)

适合:多步推理、需要“先想再做”的任务(如写报告、数据分析)

# agent_react.py
from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.prompts import PromptTemplate
from llm import llm
from tools import tools

# ReAct 提示词(必须包含 "Thought:" "Action:" "Observation:")
prompt = PromptTemplate(
    input_variables=["input", "agent_scratchpad"],
    template="""
你是一个会思考、会使用工具的智能助手。
请严格按以下格式回复:
Thought: 你当前的思考
Action: 要调用的工具名
Action Input: 工具参数
Observation: 工具返回结果
...(循环)
Final Answer: 最终答案

工具列表:
{tools}

用户问题:{input}
{agent_scratchpad}
"""
)

agent = create_react_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors="返回'思考错误'")

# 测试
if __name__ == "__main__":
    res = agent_executor.invoke({"input": "上海天气如何?计算 9876*5432"})
    print(res["output"])

加入记忆(支持多轮对话)

1. 短期记忆(会话记忆)

from langchain_core.messages import SystemMessage
from langchain.memory import ConversationBufferMemory

memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)

# 修改提示词,加入 chat_history
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是专业助手,用中文回答。"),
    ("user", "{input}"),
    ("placeholder", "{chat_history}"),
    ("agent_scratchpad", "{agent_scratchpad}")
])

# 创建带记忆的 Agent
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, memory=memory)

# 多轮测试
agent_executor.invoke({"input": "北京天气?"})
agent_executor.invoke({"input": "刚才问的是哪个城市?"})  # 能记住上下文

2. 长期记忆(向量库,RAG)

可接入 Chroma/FAISS,实现“文档问答+工具调用”的 RAG Agent,适合知识库场景。


生产级:用 LangGraph 构建复杂智能体

适合:长流程、多步骤、需要规划/反思/状态管理的任务(如自动写周报、数据处理流水线)

核心概念

  • State:全局状态(用户输入、中间结果、工具返回、记忆)
  • Nodes:思考节点、工具调用节点、总结节点
  • Edges:条件路由(是否需要继续调用工具)

极简示例(伪代码)

from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class AgentState(TypedDict):
    input: str
    messages: List
    tool_calls: List

# 1. 思考节点:LLM 决定下一步
def think_node(state: AgentState):
    res = llm.invoke(state["messages"])
    return {"tool_calls": res.tool_calls}

# 2. 工具节点:执行工具
def tool_node(state: AgentState):
    results = []
    for call in state["tool_calls"]:
        tool = next(t for t in tools if t.name == call["name"])
        results.append(tool.invoke(call["args"]))
    return {"messages": state["messages"] + results}

# 3. 构建图
graph = StateGraph(AgentState)
graph.add_node("think", think_node)
graph.add_node("tool", tool_node)
graph.add_edge("think", "tool")
graph.add_edge("tool", "think")  # 循环思考-工具
graph.set_entry_point("think")

# 编译运行
app = graph.compile()
app.invoke({"input": "复杂多步任务..."})

DeepSeek 智能体常见坑与优化

1. 工具调用不稳定(不按格式输出)

  • 降低 temperature 到 0~0.2
  • System Prompt 明确强制格式(如“必须返回 JSON”)
  • 用 handle_parsing_errors 捕获异常并重试

2. 长文本截断

  • 用 deepseek-chat(上下文 32k)
  • 开启流式输出 stream=True

3. Reasoner 模型使用

  • deepseek-reasoner 推理更强,但不支持工具调用(2026.5)
  • 组合策略:chat 做工具调用,reasoner 做深度分析

4. 安全

  • 工具参数校验(防止注入)
  • 限制 eval 等危险函数(示例仅用于演示)