28. LangChain Agent 工具调用：把手写 Function Calling Loop 交给框架#

1
agent = create_agent(model=llm, tools=[search_project_knowledge])

1
先装配机器：
2
模型用 llm
3
工具有 search_project_knowledge
4
装配结果叫 agent

1
result = agent.invoke({"messages": [{"role": "user", "content": question}]})

1
create_agent 是创建工具型助手；agent.invoke 是把用户问题交给这个助手执行。

1
chain_with_history = RunnableWithMessageHistory(...)
2
  -> 先组装一个带记忆能力的 chain
3

4
chain_with_history.invoke(...)
5
  -> 再启动一次真实调用

1
agent = create_agent(...)
2
  -> 先组装一个带工具能力的 Agent
3

4
agent.invoke(...)
5
  -> 再启动一次真实调用

1
"退款需要几天内申请？"

1
{"messages": [{"role": "user", "content": question}]}

1
HumanMessage：用户问题
2
AIMessage：模型提出 tool_call
3
ToolMessage：工具执行结果
4
AIMessage：模型最终回答

1
一个 state 字典，里面有 messages。

1
工具参数仍然要限制
2
高风险动作仍然要鉴权
3
工具返回仍然要控制长度
4
不能把敏感数据直接塞给模型

1
@tool
2
def search_project_knowledge(query: str, limit: int = 3) -> str:
3
    """Search the local project knowledge base for relevant document chunks."""
4
    docs = search_knowledge_base(query=query, limit=limit)
5
    return "..."

1
# 这是普通注释，Python 和 LangChain 不会把它当成工具说明

1
"""这是函数说明，放在函数体第一行。"""

1
print(search_project_knowledge.__doc__)

1
# 注释：给人看
2
"""docstring"""：给人看，也会被 @tool 用来生成工具说明

1
"""Search."""

1
"""Search the local project knowledge base for relevant document chunks."""

1
当用户问项目知识库里的内容时，可以调用这个工具。

1
TOOLS = [
2
    {
3
        "name": "search_knowledge_base",
4
        "description": "在知识库中搜索相关文档切片，并返回结果列表。",
5
        "parameters": {
6
            "type": "object",
7
            "properties": {
8
                "query": {"type": "string"},
9
                "limit": {"type": "integer"},
10
            },
11
            "required": ["query"],
12
        },
13
    }
14
]

1
@tool
2
def search_project_knowledge(query: str, limit: int = 3) -> str:
3
    """Search the local project knowledge base for relevant document chunks."""
4
    ...

1
手写 TOOLS JSON 是显式写说明书；@tool 是让 LangChain 根据函数自动整理说明书。

1
工具函数可以在内部使用 Document，但给模型的 tool output 最好是短、清楚、可引用的字符串。

1
import os
2

3
from dotenv import load_dotenv
4
from langchain_deepseek import ChatDeepSeek
5

6

7
load_dotenv(dotenv_path=".env")
8

9
llm = ChatDeepSeek(
10
    model=os.getenv("MODEL_NAME", "deepseek-ai/DeepSeek-V3.2"),
11
    api_base=os.getenv("MODEL_API_URL", "https://api-inference.modelscope.cn/v1"),
12
    api_key=os.getenv("MODELSCOPE_API_KEY"),
13
    temperature=0.2,
14
    streaming=False,
15
)

1
准备一个可以被 LangChain Agent 调用的模型对象。

1
模型名称、请求地址、API Key 都要先准备好。

1
from langchain.tools import tool
2

3
from app.tools.knowledge_base import search_knowledge_base
4

5

6
@tool
7
def search_project_knowledge(query: str, limit: int = 3) -> str:
8
    """Search the local project knowledge base for relevant document chunks."""
9
    safe_limit = max(1, min(limit, 5))
10
    docs = search_knowledge_base(query=query, limit=safe_limit)
11

12
    if not docs:
13
        return "没有检索到相关知识库内容。"
14

15
    formatted_docs = []
16

17
    for index, doc in enumerate(docs, start=1):
18
        metadata = doc.metadata or {}
19
        title = metadata.get("title", "未命名")
20
        source = metadata.get("source", "未知来源")
21
        chunk_index = metadata.get("chunk_index", "未知切片")
22
        content = doc.page_content[:800]
23

24
        formatted_docs.append(
25
            f"[{index}] 来源：《{title}》（{source}，chunk={chunk_index}）\n{content}"
26
        )
27

28
    return "\n\n".join(formatted_docs)

1
模型给的 arguments 可能格式对、字段对，但值没轻没重。

1
safe_limit = max(1, min(limit, 5))

1
最少 1 条，最多 5 条。

1
from langchain.agents import create_agent
2

3

4
agent = create_agent(
5
    model=llm,
6
    tools=[search_project_knowledge],
7
    system_prompt=(
8
        "你是一个严谨的知识库助手。"
9
        "如果问题需要项目知识库证据，先调用 search_project_knowledge。"
10
        "回答时说明依据来自工具返回的内容；如果没有证据，就明确说没有检索到。"
11
    ),
12
)

1
result = agent.invoke(
2
    {
3
        "messages": [
4
            {
5
                "role": "user",
6
                "content": "退款需要几天内申请？",
7
            }
8
        ]
9
    }
10
)
11

12
answer = result["messages"][-1].content
13
print(answer)

1
agent.invoke(...) 返回的不是一个字符串。

1
result["messages"]

1
result["messages"][-1].content

1
user_id = "user-1"
2
  -> thread_id = "chat-001"
3
  -> thread_id = "chat-002"

1
用户是谁，和当前是哪段对话，不是同一个问题。

1
create_agent(..., tools=...) 里优先放真实可执行的工具函数或 @tool 包装后的 Tool，不是放第 26 章那个 TOOLS 说明书列表。

1
TOOLS = [...]
2
TOOL_FUNCTIONS = {
3
    "search_knowledge_base": search_knowledge_base,
4
}

1
agent = create_agent(
2
    model=llm,
3
    tools=[search_project_knowledge],
4
)

1
给模型生成工具说明
2
保留后端可执行函数

1
# 第 26 章手写 loop 用
2
TOOLS = [...]
3

4
# 第 28 章 LangChain Agent 用
5
tools=[search_project_knowledge]

1
agent = create_agent(model=llm, tools=TOOLS)

1
工具说明书
2
真实可执行函数
3
框架包装后的 Tool

1
@tool
2
def search_project_knowledge(...):
3
    ...
4

5
agent = create_agent(..., tools=[search_project_knowledge])

1
registry.py 让你看清底层；@tool 让你进入框架写法。

1
agent = create_agent(
2
    model=____,
3
    tools=[____],
4
    system_prompt="____",
5
)
6

7
result = agent.invoke(
8
    {"messages": [{"role": "user", "content": ____}]}
9
)
10

11
answer = result["messages"][-1].content

1
llm：模型层
2
tool：工具层
3
system_prompt：行为规则层
4
user content：当前用户输入
5
最后一条 message：最终回答

1
answer = agent.invoke(...)

1
result = agent.invoke(...)
2
answer = result["messages"][-1].content

1
我用了 Agent，所以它应该天然记忆。

1
Agent 也需要 checkpointer + thread_id 才能恢复同一段对话。

1
return "\n".join(doc.page_content for doc in docs)

1
content = doc.page_content[:800]

1
return "\n\n".join(
2
    f"[{index}] {doc.page_content[:800]}"
3
    for index, doc in enumerate(docs, start=1)
4
)

1
把多个 Document，变成一个给模型看的字符串。

1
docs = [doc_a, doc_b, doc_c]

1
1, doc_a
2
2, doc_b
3
3, doc_c

1
f"[{index}] {doc.page_content[:800]}"

1
[1] 第一个切片的正文，最多取前 800 个字符
2
[2] 第二个切片的正文，最多取前 800 个字符
3
[3] 第三个切片的正文，最多取前 800 个字符

1
"\n\n".join(...)

1
[1] 第一个切片的正文，最多取前 800 个字符
2

3
[2] 第二个切片的正文，最多取前 800 个字符
4

5
[3] 第三个切片的正文，最多取前 800 个字符

1
formatted_docs = []
2

3
for index, doc in enumerate(docs, start=1):
4
    text = f"[{index}] {doc.page_content[:800]}"
5
    formatted_docs.append(text)
6

7
return "\n\n".join(formatted_docs)

1
f"[{index}] {doc.page_content[:800]}"

1
`[${index}] ${doc.pageContent}`

1
f"[{index}] {doc.page_content}"

1
字符串前面加 f
2
花括号 {...} 里面可以写变量或表达式
3
Python 会先计算 {...} 里的值
4
再把结果塞回字符串

1
index = 1
2
content = "秦始皇是秦朝的建立者"
3

4
text = f"[{index}] {content}"
5

6
print(text)

1
[1] 秦始皇是秦朝的建立者

1
f"[{index}] {doc.page_content[:800]}"

1
先算 {index}
2
  -> 比如 1
3

4
再算 {doc.page_content[:800]}
5
  -> 比如 "秦始皇是秦朝的建立者..."
6

7
最后拼成
8
  -> "[1] 秦始皇是秦朝的建立者..."

1
从 doc 的 page_content 字段里，取前 800 个字符。

1
enumerate 负责编号；f-string 负责拼每条文本；join 负责把多条文本拼成一个工具输出字符串。

1
"""Search."""

1
"""Search the local project knowledge base for relevant document chunks."""

1
一个用户永远只有一个 thread_id。

1
user_id 表示人；thread_id 表示这个人的某一条对话。

1
用了 create_agent 就不用校验工具参数了。

1
模型给参数，后端仍然要限制范围、权限和返回内容。

1
LangChain Agent 把模型、工具和执行循环组合起来，让模型可以请求工具，后端执行工具，再让模型基于工具结果回答。

1
它减少手写 Function Calling loop 的重复工作，让我不用每次都手动解析 tool_call、调用函数、追加 tool output 和再次请求模型。

1
如果我要完全控制每一步，用手写 loop 更清楚；如果我要快速组合多个工具、记忆和模型，用 LangChain Agent 更省事。

1
app/tools/knowledge_base.py 里有真实搜索函数 search_knowledge_base
2
app/tools/registry.py 是手写 Function Calling 的工具注册表
3
LangChain Agent 版本会用 @tool 包装搜索函数，再传给 create_agent(..., tools=[...])
4
最终回答从 result["messages"][-1].content 获取