Build Agent Syetem with Langgraph & DeepSeek

Environment

首先使用 pip 安装相关依赖包：

    pip install langchain
    pip install langgraph

配置一下 API 和 URL：

from dotenv import load_dotenv

import os

  

load_dotenv(".env")

  

BASE_URL = "https://api.deepseek.com"

API_KEY = os.environ.get("API_KEY")

  

deepseek_chat_model = "deepseek-chat"

Initialization Agent

一共有三种方式来初始化 Agent：

# 1. langchain: init model,invoke,prompt template（比较简单通用）
# 2. langgraph memory
# 3. langgraph create_react_agent

Langchain

DeepSeek 兼容 openai sdk，因此可以调用 langchain_openai 来导入模型。不过 DeepSeek 自己也有 langchain 专门的一个 sdk，也可以直接调用。这种方法就是 ChatXXX Model 来进行一个初始化。

from langchain_openai import ChatOpenAI

# from langchain_deepseek import ChatDeepSeek

from langchain_core.messages import HumanMessage

from langchain_core.prompts import ChatPromptTemplate

  

llm1 = ChatOpenAI(model=deepseek_chat_model, api_key=API_KEY, base_url=BASE_URL)

  

result = llm1.invoke([HumanMessage("给我讲个笑话吧")])

print(result.content)

更通用的一个方法是使用更统一的 langchain.chat_models 导入：

from langchain.chat_models import init_chat_model

  

llm2 = init_chat_model(

    deepseek_chat_model,

    api_key=API_KEY,

    base_url=BASE_URL

)

  

result = llm2.invoke([HumanMessage("给我讲个笑话吧")])

print(result.content)

初始化模型后，我们接下来可以去创建提示词模版。使用提示词模版的好处在于，每次输入的时候并不需要都传一个 message 参数，大部分时候我们的输入都只是改变其中的一部分内容，所以可以对提示词模版中的部分元素进行一个变量替换。通过方括号 “{}” 可以在 template 里替换掉对应的部分。

这个提示词模版可以用 langchain 里通用的调用接口 invoke() 来调用。针对 llm 的 invoke 是调用 AI 服务，发送消息到 AI 模型，接收回复并返回结果。而提示词模版对象的调用是文本处理，不涉及 AI 调用。

当然，invoke() 也支持一些其它对象的调用，比如 tools，chains，而这就是 langchain 的统一对象接口调用理念的体现。

使用提示词模版的时候，我们可以直接利用 langchain 里的 chain 来进行模版填充和模型回复。

这里直接构造 chain = prompt | llm1，所起到的作用就是对 | 的两端同时调用 invoke() 方法。得益于 invoke() 对不同对象的适用性，这就使得 langchain 可以将多个模型、工具、提示词串联，来构造 agent 工作流。

# 构建prompt template

from langchain_core.prompts import ChatPromptTemplate

  

prompt = ChatPromptTemplate(

    [

        ("system", "你是一个笑话大王，请给我讲个笑话"),

        ("user", "帮我写一个主题为&#123;topic&#125;的笑话")

    ]

)

  

prompt.invoke(&#123;"topic": "狗"&#125;).to_messages()

print(prompt.input_variables)

  

# langchian里chain的体现

# | 连接的两侧必须是可以被invoke的，在调用时会自动调用两侧的invoke方法

# 左边的输出会作为右边的输入

chain = prompt | llm1

  

ans = chain.invoke(

    &#123;"topic": "狗"&#125;

)

  

print(ans.content)

生成的笑话，虽然有点冷，，

补充：事实上一般在我们去做一些文本生成、数据生成等任务的时候，我们可能更希望有一个 json 格式的输出结果。如何让模型的输出结果自动返回 json 格式呢？

首先，我们可以借助 Pydantic 库，定义一个匹配我们需求的输出 json 格式的数据结构。Pydantic 的 BaseModel 类用于进行数据验证、类型转换等操作，确保返回格式符合需求的 json 格式要求；
接下来，我们利用模型 API 的 with_structured_output() 输出方法，将 Pydantic 模型转换为 JSON Schema 并使用 Function Calling 机制，在提示词中嵌入结构要求，令模型的输出结果符合所需的 json 格式。

from pydantic import BaseModel, Field

  

class Joke(BaseModel):

    "joke to tell user"

    setup: str = Field(description="The setup of the joke")

    punchline: str = Field(description="The punchline of the joke")

  

joke_chain = prompt | ChatDeepSeek(

    model=deepseek_chat_model,

    api_key=API_KEY,

    base_url=BASE_URL

).with_structured_output(Joke)

  

ans = joke_chain.invoke(&#123;"topic": "狗"&#125;)

print(ans.setup)

print(ans.punchline)

注意，这里的方法仅适用于支持 Function Calling 的模型！

因为实际上我们采用 with_structured_output(class) 来让模型输出 json 格式时，其实是虚拟了一个自定义的 function，这个 function 规定了一个 json 转换的模式，在调用 API 的时候会打开 function calling 采用这个虚拟 function，来实现 json 格式的转换，并通过 pydantic 库验证输出。

function calling 就是自定义函数支持。简单来说，用户可以预先定义好函数（比如天气获取函数），模型可以选择调用函数，执行函数并返回结果，最终模型基于函数结果生成回答。

函数信息可以通过每次调用 API 时传递，模型在收到这样的消息，判断要调用一些函数，便不生成实际回复，而是生成一个函数调用并结束该次响应。应用程序执行函数 → 再次调用模型 → 模型基于结果生成最终回答。

当然也有模型直接支持启用 json 模式，这时候生成结果就是 json 格式的：

llm_with_json = ChatOpenAI(

    model="gpt-4-1106-preview",  # 支持 JSON Mode

    response_format=&#123;"type": "json_object"&#125;

)


result = llm_with_json.invoke([HumanMessage("生成笑话")])

Langgraph Memory

langgraph 基于图结构来编排 Agent 工作流，其关键点就在于图结构的 message memory 储存机制。LangGraph的核心理念是将应用逻辑表示为一个图（Graph），其中：

节点（Nodes）：代表工作流中的独立组件或智能体，可以视为以特定方式相互交互的”参与者”。每个节点可以执行特定的功能，如生成文本、分析数据或调用外部工具。
边（Edges）：是Python中的函数，基于当前状态决定下一个要执行的节点，实现了工作流中的条件逻辑和控制流。
状态（State）：作为一种”记忆库”，记录和跟踪AI系统处理的所有有价值的信息，确保系统能够维持上下文并做出连贯的响应。

图驱动的架构使 langgraph 在需要上下文连贯性的应用中表现比较出色。

# memory: message & output, 通过维护图结构来保存

  

# class MessagesState(TypedDict):

#     messages: Annotated[list[AnyMessage], add_messages]

#     &#123;

#         "messages": []

#     &#125;

通过维护消息状态，langgraph 会通过合并消息、新增消息节点等来进行状态管理。

这其实就是一个图结构的 Agent Memory，实现的是一个短期记忆。对话消息上下文会通过图结构的维护来传递给 Agent。

from langgraph.checkpoint.memory import MemorySaver

from langgraph.graph import MessagesState, START, StateGraph, END

  

workflow = StateGraph(MessagesState)

  

def call_model(state: MessagesState):

    return &#123;"messages": [llm1.invoke(state["messages"])]&#125;

  

workflow.add_node("node1", call_model)

workflow.add_edge(START, "node1")

  

# Add Memory

memory = MemorySaver()

graph = workflow.compile(checkpointer=memory)

  

# config标识线程id，是通过线程id来存储历史消息的

config = &#123;"configurable": &#123;"thread_id": "1"&#125;&#125;

query = "你好啊，我是Cyan。"

  

input_messages = [HumanMessage(content=query)]

output_messages = graph.invoke(input_messages, config=config)

print(output_messages["messages"][-1].content)

# 这里的output包含所有历史消息，langgraph做了一个消息拼接，[-1] 获取最后一条消息（最新的AI回复）

  

query = "我叫什么名字？"

input_messages = [HumanMessage(content=query)]

output_messages = graph.invoke(input_messages, config=config)

print(output_messages["messages"][-1].content)

Single Agent (with tools)

使用 function calling 来 create 一个 ReAct 的单 agent 系统，agent 能够在推理过程中去调用已有声明的自定义函数帮助完成推理过程：

from langgraph.prebuilt import create_react_agent

  

def add(a:int, b:int):

    return a + b

  

def multiply(a:int, b:int):

    return a * b

  

agent = create_react_agent(

    model=llm1,

    tools=[add, multiply],

    config=&#123;"configurable": &#123;"thread_id": "1"&#125;&#125;

)

  

input_messages = [HumanMessage(content="计算10+20，然后计算10*20")]

output_messages = agent.invoke(input_messages)

  

for message in output_messages["messages"]:

    print(message.content)