上一篇文章一文弄懂Agent从手搓Claude开始，我们介绍了如何手搓Agent，发现有些事情必须要去做：比如和LLM传递信息，使用Tools，用ReAct增强Agent，用RAG获取外部信息，记住历史信息等等。我曾经说过，重复的事情做三遍，就应该考虑工具化。

LangChain正是在这样的背景下诞生的，它提供了一个辅助实现这些功能的框架和工具集，从而帮助开发者构建更加强大，可落地的商业Agent应用程序。本文会探讨LangChain的基础核心功能，以及我自己对框架使用的思考。

1. 准备工作
   =======

2. 安装langchain

pip install -U langchain==1.0# Requires Python 3.10+

2. 使用deepseek-chat作为后端大模型

llm = init_chat_model("deepseek-chat",    api_key="<API KEY>",    temperature=0.5,    timeout=20,    max_tokens=1000,    http_client=httpx.Client(verify=False),)

3. 打印agent和大模型通信的request，response，为了理解langchain背后的原理，查看通信内容会有很大帮助。

def log_request(request):
print("============ Outgoing Request ============")
print(f"Method: {request.method}")
print(f"URL: {request.url}")
# 记录请求体
if hasattr(request, 'content'):
try:
if isinstance(request.content, (bytes, bytearray)):
body = request.content.decode('utf-8')
else:
body = str(request.content)
try:
body_json = json.loads(body)
print("Body (JSON):")
print(json.dumps(body_json, indent=2, ensure_ascii=False))
except:
print(f"Body: {body}")
except Exception as e:
print(f"Body: [Unable to read: {e}]")
print("===============================")
def log_response(response):
print("=========== Incoming Response ===========")
print(f"Status: {response.status_code}")
# 读取响应内容
try:
response.read()  # 这步很重要！
content = response.text
# 尝试美化JSON输出
try:
content_json = json.loads(content)
print("Content (JSON):")
print(json.dumps(content_json, indent=2, ensure_ascii=False))
except:
# 限制输出长度
if len(content) > 1000:
print(f"Content: {content[:1000]}...")
else:
print(f"Content: {content}")
except Exception as e:
print(f"Error reading response content: {e}")
print("==============================")
# 创建自定义 HTTP client
http_client = httpx.Client(
verify=False,
event_hooks={
'request': [log_request],
'response': [log_response],
}
)
llm = init_chat_model(
"deepseek-chat",
api_key="<API KEY>",
temperature=0.5,
timeout=20,
max_tokens=1000,
http_client=http_client,
)
agent = create_agent(
model=llm,
system_prompt="You are a helpful assistant",
)

2. Messages
   ===========

Messages是LangChain里最基础的概念，代表了对模型的输入和输出。一个典型的使用如下：

from langchain.messages import SystemMessage, HumanMessage, AIMessagemessages = [    SystemMessage("You are a poetry expert"),    HumanMessage("Write a haiku about spring"),    AIMessage("Cherry blossoms bloom...")]response = llm.invoke(messages)

实际上这些不同的Messages类型就对应System、User、Assistant这三个role，不用对象封装，手搓的话，与下面的效果等价

messages = [    {"role": "system", "content": "You are a poetry expert"},    {"role": "user", "content": "Write a haiku about spring"},    {"role": "assistant", "content": "Cherry blossoms bloom..."}]response = llm.invoke(messages)

3. Tools
   ========

使用工具是Agent的基本能力，在langchain的封装下，我们不再需要自己写使用工具的prompt，而是用如下的方式：

from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from deepseek_model import llm, log_before_model
# Access the current conversation state
@tool
def summarize_conversation(
runtime: ToolRuntime
) -> str:
"""Summarize the conversation so far."""
messages = runtime.state["messages"]
human_msgs = sum(1 for m in messages if m.__class__.__name__ == "HumanMessage")
ai_msgs = sum(1 for m in messages if m.__class__.__name__ == "AIMessage")
return f"Conversation has {human_msgs} user messages, {ai_msgs} AI responses"
agent = create_agent(
model=llm,
system_prompt="You are a helpful assistant",
tools=[summarize_conversation],
middleware=[log_before_model],
)
messages = {"messages": [
{"role": "system", "content": "when the number of messages is asked, you can call summarize_conversation tool"},
{"role": "user", "content": "Nice to meet you"},
{"role": "assistant", "content": "Nice to meet you too"},
{"role": "user", "content": "tell me how many user and assistant messages are in this conversation"}
]}
response = agent.invoke(messages, print_mode="values")
for m in response["messages"]:
if m.__class__.__name__ == "AIMessage":
print(m.content)

其背后，langchain是把通过@tool标识的工具类，用tools=[summarize_conversation]注册到agent。原理就是通过python的Decorator把函数变成StructuredTool类：

class StructuredTool(BaseTool):
"""Tool that can operate on any number of inputs."""
description: str = ""
args_schema: Annotated[ArgsSchema, SkipValidation()] = Field(
..., description="The tool schema."
)
"""The input arguments' schema."""
func: Callable[..., Any] | None = None
"""The function to run when the tool is called."""
coroutine: Callable[..., Awaitable[Any]] | None = None
"""The asynchronous version of the function."""

这个类的主要目的是方便agent把函数信息转化为llm可以理解的工具使用的Prompt，这些内容包括函数的name、description，以及args_schema（参数信息）。通过打印给发送给llm的request内容，可以发现，生成的关于tool使用的prompt如下所示：

"tools": [
{
"type": "function",
"function": {
"name": "summarize_conversation",
"description": "Summarize the conversation so far.",
"parameters": {
"properties": {},
"type": "object"
}
}
}
]

llm返回给agent的response内容如下，这是在告诉agent要去执行tool call：

"message": {
"role": "assistant",
"content": "I'll check how many messages are in this conversation for you.",
"tool_calls": [
{
"index": 0,
"id": "call_00_GhtKT5QA8SmSLq9GcsNpDlbq",
"type": "function",
"function": {
"name": "summarize_conversation",
"arguments": "{}"
}
}
]
},
"logprobs": null,
"finish_reason": "tool_calls"
}

对比一文弄懂Agent从手搓Claude开始中我们实现Function Calling的方式，不同点只是使用的格式（或者说我们和LLM之间的协定）不一样，但原理都是一样的。

4. Memory
   =========

记忆（Memory）是一个能够记录以往交互信息的系统。对于智能体而言，记忆能力至关重要——它不仅能保存历史交互记录，更能从反馈中学习进化，逐步适应不同用户的个性化需求。随着智能体需要处理愈发复杂的任务和海量用户交互，这种记忆能力已成为提升运行效率和用户体验的核心要素。

我们以long-term memory为例，演示一下其使用情况，假设这些信息要存入postgres数据库。

from dataclasses import dataclass
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from langgraph.store.postgres import PostgresStore  # 改为 PostgreSQL 存储
from typing_extensions import TypedDict
from deepseek_model import llm
DB_URI = "postgresql://postgres:1314520@localhost:5432/Test?sslmode=disable"
with PostgresStore.from_conn_string(
conn_string=DB_URI,
pipeline=False,
pool_config=None,
index=None,
ttl=None
) as store:
store.setup()
@dataclass
class Context:
user_id: str
# TypedDict defines the structure of user information for the LLM
class UserInfo(TypedDict):
name: str
age: int
address: str
# Tool that allows agent to update user information (useful for chat applications)
@tool
def save_user_info(user_info: UserInfo, runtime: ToolRuntime[Context, dict]) -> str:
"""Save user info."""
# Access the store - same as that provided to `create_agent`
store = runtime.store
user_id = runtime.context.user_id
# Store data in the store (namespace, key, data)
store.put(("users",), user_id, user_info)
return "Successfully saved user info."
agent = create_agent(
model=llm,
tools=[save_user_info],
store=store,
context_schema=Context
)
# Run the agent
result = agent.invoke(
{"messages": [{"role": "user", "content": "My name is Frank Smith, 48 years old, live at 1024 Hamilton Ave. in San Jose, California. please save it using save_user_info function"}]},
# user_id passed in context to identify whose information is being updated
context=Context(user_id="user_123")
)
for m in result["messages"]:
if m.__class__.__name__ == "AIMessage":
print(m.content)
# You can access the store directly to get the value
user_info = store.get(("users",), "user_123")
print(user_info)

运行完以上程序，我们会发现在数据库中的store表中，多出了一条记录，且这条记录可以随时从db中取出。通过数据库的持久化，我们就可以长时间保留用户信息。在商业应用中，我们的Agent服务肯定都是多实例的，这种持久化也能实现服务的stateless，从而让用户的request可以在不同的Agent实例间迁移。

5. Structured output
   ====================

Structured output使Agent能够以特定、可预测的格式返回数据。LangChain的create_agent可自动处理结构化输出：用户设定所需的结构化输出schema后，当模型生成结构化数据时，系统会自动捕获并验证这些数据，最终通过dict的’structured_response’键返回处理结果。

以下是个简单的示例：

from typing import Literal
from langchain.agents import create_agent
from pydantic import BaseModel, Field
from deepseek_model import llm, print_dict_structured
class ProductReview(BaseModel):
"""Analysis of a product review."""
rating: int | None = Field(description="The rating of the product", ge=1, le=5)
sentiment: Literal["positive", "negative"] = Field(description="The sentiment of the review")
key_points: list[str] = Field(description="The key points of the review. Lowercase, 1-3 words each.")
agent = create_agent(
model=llm,
response_format=ProductReview
)
result = agent.invoke({
"messages": [{"role": "user",
"content": "Analyze this review: 'bad product: 2 out of 5 stars. Fast shipping, but too expensive'"}]
})
review = result["structured_response"]
print(review)

执行上面的程序，会得到以下结果：

'structured_response': ProductReview(rating=2, sentiment='negative', key_points=['fast shipping', 'too expensive']

其背后原理，也是通过Function Calling实现的，观察agent发送给llm的request日志，会看到以下的内容，就是一个典型的Function Calling 的prompt

"tools": [
{
"type": "function",
"function": {
"name": "ProductReview",
"description": "Analysis of a product review.",
"parameters": {
"properties": {
"rating": {
"anyOf": [
{
"maximum": 5,
"minimum": 1,
"type": "integer"
},
{
"type": "null"
}
],
"description": "The rating of the product"
},
"sentiment": {
"description": "The sentiment of the review",
"enum": [
"positive",
"negative"
],
"type": "string"
},
"key_points": {
"description": "The key points of the review. Lowercase, 1-3 words each.",
"items": {
"type": "string"
},
"type": "array"
}
},
"required": [
"rating",
"sentiment",
"key_points"
],
"type": "object"
}
}
}
]

6. 链式处理
   =======

LCEL（LangChain Expression Language）也是Chain这个名字的来历，LCEL语法利用了 Python 的 运算符重载 特性，具体来说是 or 方法重载。从而使得运算符|有了类似于管道的功能，可以串联一系列Agent需要的操作，比如基本示例：提示 + 模型 + 输出解析器：

prompt = ChatPromptTemplate.from_template("请用中文回答：{question}")
output_parser = StrOutputParser()
chain = prompt | llm | output_parser
print("begin invoke")
try:
result = chain.invoke({"question": "杭州今天的天气怎么样？"})
print(result)
except Exception as e:
print(f"调用失败: {e}")

其背后的原理，就是运算符重载，以下是一个模拟LangChain的LCEL语法实现：

class Runnable:
def __init__(self, name):
self.name = name
def __or__(self, other):
# 模拟 LCEL 的链式连接
return Chain([self, other])
def run(self, input_data):
return f"{self.name}处理({input_data})"
def __str__(self):
return self.name
class Chain:
def __init__(self, steps):
self.steps = steps
def run(self, input_data):
result = input_data
for step in self.steps:
result = step.run(result)
return result
def __or__(self, other):
return Chain(self.steps + [other])
# 使用 | 操作符构建处理链
step1 = Runnable("加载数据")
step2 = Runnable("清洗数据")
step3 = Runnable("分析数据")
pipeline = step1 | step2 | step3
result = pipeline.run("原始数据")
print(result)  # 分析数据处理(清洗数据处理(加载数据处理(原始数据)))

但是在LangChain1.0之后，对于更加复杂的场景，比如复杂的状态管理、多分支逻辑、循环或涉及多个代理协作时，更加推荐使用LangGraph来处理。

7. LangChain中的软件设计
   ==================

作为一个软件老登，在关注LangChain的功能的同时，一定也会关注它是怎么设计和实现的。我发现有两个典型的设计：一个是上下文设计，另一个是可扩展设计。在langchain中也有用到

7.1 上下文设计

Context在框架中之所以重要，是因为我们在处理信息的时候，往往要借助上下文信息。比如在java的web框架中有ServletContext，在Spring框架中有ApplicationContext。Context进一步可以分为Procedure Context（过程上下文）和Global Context（全局上下文）：

在langchain中，上下文主要是通过ToolRuntime实现的：

@dataclass
class ToolRuntime
state: StateT
context: ContextT
config: RunnableConfig
stream_writer: StreamWriter
tool_call_id: str | None
store: BaseStore | None

结合Memory的内容，我们可以将langchain中的context分为以下三类：

类别	别称	作用范围	示例
Context	静态配置	会话范围	用户ID、API密钥、数据库连接、权限、环境设置
State	短期记忆	会话范围	当前消息、上传文件、认证状态、工具执行结果
Store	长期记忆	跨会话	用户偏好、提取的洞察、记忆信息、历史数据

其中，Context是静态信息，属于Global Context。Sate、Store是Procedure Context。准确的说Store不仅仅是single procedure的，也可以是cross-procedure的。

7.2 可扩展设计

• Middleware扩展

  Middleware在python的语境下，是一种常用的设计模式，和过滤器、管道模式类似。通过Middleware我们可以在横切面添加额外的功能。

客户端请求 → Middleware1 → Middleware2 → ... → 业务处理 → Middleware2 → Middleware1 → 客户端响应

Langchain除了大量的build-in的middleware，用户可以在以下的扩展点（hook点）自定义middleware，从而极大的提升了框架的可扩展性，基本上，所有的框架都有类似的设计。

• 依赖倒置

  langchain中的Store是抽象的，这个设计和我在cola-job中的设计是一模一样的，通过依赖倒置，磨平具体实现InMemoryStore，DBStore，RedisStore之间的差异。有时候，我们也把这个模式叫面向接口编程。

• 泛型

  Python在3.5之后引入了泛型。（我个人更喜欢强类型的语言，这样很多类型问题编译时就能提示）在上面的例子中，我们已经看到了用户自定义State、Context的方法。实际上都是因为StateT和ContextT泛型的使用，从而在保证扩展性的同时，同时兼顾了类型安全。

8. 关于框架的反思
   ==========

从上文的原理解释中，你应该也发现了，使用LangChain构建Agent和手搓Agent的最大区别是，LangChain通过抽象、封装的方式，帮我们生成了本来需要我们自己写的Prompt，以及和LLM的交互细节也被屏蔽了，比如Function Calling的过程。在带来了一定便利性的同时，也给像我这样的初学者带来了不少困惑，很多事情没有像手搓Agent时那么直观，要通过看日志、看它的源码，才能理解其背后的原理。

这是一个典型的使用框架带来的问题，即框架作为一个“黑盒”，会增加学习成本，性能成本，debug成本，维护成本等。使用不当，还会额外引入复杂度，我曾经猛烈的抨击过流程引擎，在我经历过的项目中，但凡引入流程引擎框架的，无一例外都被搞得乌烟瘴气，得不偿失。本来简单直接的业务逻辑，在流程引擎的编排下，变得支离破碎，上下文不连续，晦涩难懂，更加复杂。

所以对于LangChain，特别是LangGraph（本质上就是一个workflow工作流编排工具），请各位一定要谨慎选择，不要轻易趟这个浑水。如果觉得我是危言耸听的，可以看看这篇文章《为什么我们不再使用 LangChain 来构建我们的 AI 智能体》

因为篇幅关系，我会在下一篇介绍LangChain的其它三个扩展模块，LangGraph，DeepAgents 和 LangSmith

如何学习AI大模型 ？

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。【保证100%免费】🆓

CSDN粉丝独家福利

这份完整版的 AI 大模型学习资料已经上传CSDN，朋友们如果需要可以扫描下方二维码&点击下方CSDN官方认证链接免费领取 【保证100%免费】

读者福利： 👉👉CSDN大礼包：《最新AI大模型学习资源包》免费分享 👈👈

（👆👆👆安全链接，放心点击）

对于0基础小白入门：

如果你是零基础小白，想快速入门大模型是可以考虑的。

一方面是学习时间相对较短，学习内容更全面更集中。
二方面是可以根据这些资料规划好学习计划和方向。

👉1.大模型入门学习思维导图👈

要学习一门新的技术，作为新手一定要先学习成长路线图，方向不对，努力白费。

对于从来没有接触过AI大模型的同学，我们帮你准备了详细的学习成长路线图&学习规划。可以说是最科学最系统的学习路线，大家跟着这个大的方向学习准没问题。（全套教程文末领取哈）

👉2.AGI大模型配套视频👈

很多朋友都不喜欢晦涩的文字，我也为大家准备了视频教程，每个章节都是当前板块的精华浓缩。

👉3.大模型实际应用报告合集👈

这套包含640份报告的合集，涵盖了AI大模型的理论研究、技术实现、行业应用等多个方面。无论您是科研人员、工程师，还是对AI大模型感兴趣的爱好者，这套报告合集都将为您提供宝贵的信息和启示。（全套教程文末领取哈）

👉4.大模型实战项目&项目源码👈

光学理论是没用的，要学会跟着一起做，要动手实操，才能将自己的所学运用到实际当中去，这时候可以搞点实战项目来学习。（全套教程文末领取哈）

👉5.大模型经典学习电子书👈

随着人工智能技术的飞速发展，AI大模型已经成为了当今科技领域的一大热点。这些大型预训练模型，如GPT-3、BERT、XLNet等，以其强大的语言理解和生成能力，正在改变我们对人工智能的认识。 那以下这些PDF籍就是非常不错的学习资源。（全套教程文末领取哈）

👉6.大模型面试题&答案👈

截至目前大模型已经超过200个，在大模型纵横的时代，不仅大模型技术越来越卷，就连大模型相关的岗位和面试也开始越来越卷了。为了让大家更容易上车大模型算法赛道，我总结了大模型常考的面试题。（全套教程文末领取哈）

为什么分享这些资料?

只要你是真心想学AI大模型，我这份资料就可以无偿分享给你学习，我国在这方面的相关人才比较紧缺，大模型行业确实也需要更多的有志之士加入进来，我也真心希望帮助大家学好这门技术，如果日后有什么学习上的问题，欢迎找我交流，有技术上面的问题，我是很愿意去帮助大家的！

这些资料真的有用吗?

这份资料由我和鲁为民博士共同整理，鲁为民博士先后获得了北京清华大学学士和美国加州理工学院博士学位，在包括IEEE Transactions等学术期刊和诸多国际会议上发表了超过50篇学术论文、取得了多项美国和中国发明专利，同时还斩获了吴文俊人工智能科学技术奖。目前我正在和鲁博士共同进行人工智能的研究。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

CSDN粉丝独家福利

这份完整版的 AI 大模型学习资料已经上传CSDN，朋友们如果需要可以扫描下方二维码&点击下方CSDN官方认证链接免费领取 【保证100%免费】

读者福利： 👉👉CSDN大礼包：《最新AI大模型学习资源包》免费分享 👈👈

（👆👆👆安全链接，放心点击）