模型能够生成回答，但任务往往需要外部资料和实际操作。知识与工具流程应明确区分提出请求、执行请求和确认结果。以回答RAG的基本流程为例，系统先取得资料，保留证据ID，再生成答案；没有资料时应明确停止。
目标、状态、执行边界和验收标准
| 项目 | 定义 |
|---|---|
| 目标 | 回答问题，并返回实际取得的证据ID。 |
| 状态 | messages记录模型与工具交互；seen记录已取得的证据；calls记录调用数；status区分完成与资料不足。 |
| 执行边界 | 只允许search_docs；工具最多调用两次，模型交互最多四轮；资料只来自本地教学集合；未知工具和未知证据ID被拒绝。 |
| 验收标准 | 先取得资料再形成答案；资料为空返回insufficient_evidence；输出证据ID必须存在于本次结果；调用数不得超过预算。 |
方法与设计思想
工具调用：把意图变成受控执行
模型返回工具名称和参数，程序检查名称、预算与参数后执行。工具结果通过ToolMessage回到消息序列，供模型继续处理。模型可以选择工具，但不能创造新的执行权限。
调用成功需要确认工具返回的实际数据。本流程取得空结果时直接进入资料不足状态，避免模型用已有常识掩盖检索失败。
RAG：先取得事实，再组织答案
RAG包含准备数据、检索资料、组织上下文和生成答案。代码使用Document保存内容与ID，用小型词项匹配实现可控检索。该方式方便验证完整数据流，不代表向量检索或混合召回的效果。
验收分两层：程序检查证据ID确实出现过；人工或语义评估检查答案中的结论是否得到相应内容支持。ID存在只能证明来源关系，不能证明结论正确。
MCP：在需要标准接入时加入适配层
本地函数足以完成主流程，因此主代码没有强行引入网络协议。需要共享工具或接入现有服务时，可以用MCP把服务工具转换为LangChain工具。后面的独立示例使用stdio连接本地服务，展示发现与调用过程。
MCP解决的是接入与能力描述。过滤、排序、返回格式和错误信息仍由工具接口设计决定。GraphRAG也只有在任务确实需要跨实体关系时才值得加入。
为什么加入这些机制
| 机制 | 解决的问题 | 验证方式 |
|---|---|---|
| 工具调用 | 取得模型内部无法直接确认的信息 | 检查工具请求、真实返回与ToolMessage是否对应 |
| RAG | 让生成结果具备本次取得的事实基础 | 空检索应停止；未知证据ID应拒绝；人工检查结论支持关系 |
| MCP适配 | 把服务工具接入统一工具接口 | 启动本地服务，确认工具发现和实际执行都成功 |
LangChain与LangGraph代码实现
主代码在Python 3.12.14、LangChain 1.4.3、langchain-core 1.6.6、LangGraph 1.2.12、Pydantic 2.13.5下运行验证。建议使用独立虚拟环境：
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install langchain==1.4.3 langchain-core==1.6.6 langgraph==1.2.12 pydantic==2.13.5
将下面完整代码保存为flow02.py，运行python flow02.py。默认模式不需要模型凭据，使用固定数据验证流程和异常分支。
import json
import os
from langchain.chat_models import init_chat_model
from langchain_core.documents import Document
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from pydantic import BaseModel, Field
DOCS = [
Document(page_content="RAG先检索外部证据，再生成答案。", metadata={"id": "d1"}),
Document(page_content="长期记忆保存跨任务的信息与经验。", metadata={"id": "d2"}),
]
@tool
def search_docs(query: str) -> str:
"""检索本地教学资料，返回证据ID和原始内容。"""
# 小型语料采用明确的词项匹配；不宣称向量召回效果。
hits = [d for d in DOCS if query.casefold() in d.page_content.casefold()]
return json.dumps([{"id": d.metadata["id"], "text": d.page_content}
for d in hits], ensure_ascii=False)
class Answer(BaseModel):
answer: str = Field(min_length=1)
evidence_ids: list[str] = Field(min_length=1)
def check_answer(text: str, seen: set[str]) -> Answer:
answer = Answer.model_validate_json(text)
if not set(answer.evidence_ids) <= seen:
raise ValueError("答案使用了未取得的证据ID")
return answer
def run(question: str, query: str = "RAG") -> dict:
messages = [HumanMessage(content=question)]
seen = set()
if os.getenv("MODEL_ID"):
from langchain_core.messages import SystemMessage
model = init_chat_model(os.environ["MODEL_ID"]).bind_tools([search_docs])
messages.insert(0, SystemMessage(content=
'先使用search_docs获取资料，只能输出有证据支持的结论。'
'最终只输出JSON：{"answer":"...","evidence_ids":["d1"]}。'
'没有证据时说明不足，不编造ID。'))
next_message = lambda: model.invoke(messages)
else:
scripted = iter([
AIMessage(content="", tool_calls=[{
"name": "search_docs", "args": {"query": query},
"id": "call-1", "type": "tool_call",
}]),
AIMessage(content=json.dumps({"answer": DOCS[0].page_content,
"evidence_ids": ["d1"]}, ensure_ascii=False)),
])
next_message = lambda: next(scripted)
calls = 0
for _ in range(4):
msg = next_message()
messages.append(msg)
if not msg.tool_calls:
answer = check_answer(msg.content, seen)
return {"status": "completed", "calls": calls,
"evidence": sorted(seen), **answer.model_dump()}
for request in msg.tool_calls:
if request["name"] != "search_docs":
raise ValueError("工具不在允许列表")
calls += 1
if calls > 2:
raise RuntimeError("工具调用预算耗尽")
content = search_docs.invoke(request["args"])
hits = json.loads(content)
if not hits:
return {"status": "insufficient_evidence", "calls": calls}
seen.update(hit["id"] for hit in hits)
messages.append(ToolMessage(content=content, tool_call_id=request["id"]))
raise RuntimeError("模型交互轮数耗尽")
if __name__ == "__main__":
result = run("RAG的基本流程是什么？")
assert result["status"] == "completed"
if not os.getenv("MODEL_ID"):
assert run("未知问题", query="不存在的词")["status"] == "insufficient_evidence"
try:
check_answer('{"answer":"无依据结论","evidence_ids":["d9"]}', {"d1"})
except ValueError:
pass
else:
raise AssertionError("无效证据未被拒绝")
print(json.dumps(result, ensure_ascii=False, indent=2))
search_docs返回证据ID和原始内容。run保留消息与证据集合，在执行工具之前检查允许列表和调用预算。check_answer负责结构和ID有效性；它不承担事实推断的最终裁判。真实模型模式使用bind_tools，让模型提出工具请求；固定模式使用明确的AIMessage工具请求复现交互过程。
MCP本地工具接入
在同一目录保存以下两个文件，另安装langchain-mcp-adapters==0.3.2与mcp==1.30.0，运行python mcp_client.py。客户端会启动本地stdio服务，不需要占用端口。
mcp_server.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("knowledge-demo")
@mcp.tool()
def explain_rag() -> str:
"""返回RAG的教学定义，不访问外部网络。"""
return "RAG通过检索外部资料为生成答案补充依据。"
if __name__ == "__main__":
mcp.run(transport="stdio")
mcp_client.py
import asyncio
import os
import sys
from pathlib import Path
from langchain_mcp_adapters.client import MultiServerMCPClient
async def main():
client = MultiServerMCPClient({"knowledge": {
"transport": "stdio", "command": sys.executable,
"args": [str(Path(__file__).with_name("mcp_server.py").resolve())],
}})
tools = await client.get_tools()
lookup = {tool.name: tool for tool in tools}
assert "explain_rag" in lookup
result = await lookup["explain_rag"].ainvoke({})
assert "RAG" in str(result)
print(result)
if os.getenv("MODEL_ID"):
from langchain.agents import create_agent
agent = create_agent(model=os.environ["MODEL_ID"], tools=tools,
system_prompt="通过工具获取资料，再回答问题。")
result = await agent.ainvoke({"messages": [
{"role": "user", "content": "解释RAG"},
]}, {"recursion_limit": 10})
print(result["messages"][-1].content)
if __name__ == "__main__":
asyncio.run(main())
已验证工具发现和直接调用。设置MODEL_ID后，客户端可以创建真实模型智能体；该分支需要另外安装所选供应商的LangChain集成包，并设置对应环境变量，未作为无凭据测试的一部分。
接入真实模型
代码通过init_chat_model读取MODEL_ID。安装所选供应商的集成包，配置其认证环境变量，再把MODEL_ID设为供应商:模型名称即可切换。供应商和模型需支持相应功能；第二篇需要工具调用能力。敏感参数放在环境变量中，不能写入正文或源码。真实模型输出允许变化，结构与边界检查保持相同。
验收与验证结果
主脚本验证正常回答、空检索和不存在的证据ID。MCP示例验证本地工具可发现、可调用。真实模型的工具选择、查询改写和答案事实质量没有通过这些固定样例得到证明，需要针对实际资料建立问题集。
Google ADK实现与LangChain对照
Google ADK版本用FunctionTool描述检索函数，再用确定性代理调用工具并保存证据。该版本验证检索与状态契约；模型自主工具选择是可加入的独立层。LangChain主版本同时展示了AIMessage工具请求与ToolMessage结果回传。
| 比较项 | LangChain／LangGraph | Google ADK |
|---|---|---|
| 工具定义 | @tool生成工具参数与描述 | FunctionTool包装普通Python函数 |
| 工具执行 | 程序校验AIMessage中的请求后invoke | 自定义Step调用FunctionTool.run_async |
| 结果保存 | messages和seen保留交互与证据ID | EventActions.state_delta提交evidence与状态 |
| 模型选择 | 真实模型模式使用bind_tools | 可使用LlmAgent(tools=[search_docs])；对照主代码为固定调用 |
Google ADK完整代码
对照代码已在google-adk==2.11.0与Python 3.12环境运行。安装python -m pip install google-adk==2.11.0，保存为adk02.py后运行。默认不调用真实模型。
import asyncio
import inspect
import json
from collections.abc import Callable
from typing import Any
from google.adk.agents import BaseAgent, SequentialAgent, ParallelAgent, LoopAgent
from google.adk.events import Event, EventActions
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
class Step(BaseAgent):
action: Callable[..., Any]
async def _run_async_impl(self, ctx):
delta = self.action(ctx.session.state)
if inspect.isawaitable(delta):
delta = await delta
yield Event(author=self.name, actions=EventActions(state_delta=delta))
async def run_state(agent, state, session_id="s1", user_id="u1", service=None):
service = service or InMemorySessionService()
await service.create_session(app_name="demo", user_id=user_id,
session_id=session_id, state=state)
runner = Runner(agent=agent, app_name="demo", session_service=service)
async for _ in runner.run_async(user_id=user_id, session_id=session_id,
new_message=types.Content(role="user", parts=[types.Part(text="执行任务")])):
pass
session = await service.get_session(app_name="demo", user_id=user_id,
session_id=session_id)
return dict(session.state)
from google.adk.tools import FunctionTool
DOCS = {"d1": "RAG通过检索外部资料补充答案依据。"}
def search_docs(query: str) -> list[dict]:
"""在本地教学资料中检索内容及证据ID。"""
return [{"id": key, "text": value} for key, value in DOCS.items()
if query.casefold() in value.casefold()]
search_tool = FunctionTool(func=search_docs)
async def retrieve(state):
hits = await search_tool.run_async(args={"query": state["query"]}, tool_context=None)
return {"evidence": hits, "calls": 1,
"status": "running" if hits else "insufficient_evidence"}
def write(state):
if state["status"] == "insufficient_evidence":
return {}
ids = [item["id"] for item in state["evidence"]]
return {"answer": "\n".join(item["text"] for item in state["evidence"]),
"evidence_ids": ids, "status": "completed"}
def build():
return SequentialAgent(name="knowledge", sub_agents=[
Step(name="retrieve", action=retrieve), Step(name="write", action=write),
])
async def main():
result = await run_state(build(), {"query": "RAG"})
assert result["status"] == "completed" and result["evidence_ids"] == ["d1"]
empty = await run_state(build(), {"query": "不存在的词"})
assert empty["status"] == "insufficient_evidence"
print(json.dumps(result, ensure_ascii=False, indent=2))
if __name__ == "__main__":
asyncio.run(main())
ADK测试覆盖有效资料和空资料停止。tool_context=None只用于这个不依赖上下文的本地纯函数，真实工具需要状态或权限时，应使用框架提供的ToolContext。两个示例均未证明答案事实质量。
接口与版本边界
对照采用SequentialAgent、ParallelAgent、LoopAgent或BaseAgent模板接口，以便比较相同控制思想。ADK 2.x官方文档已引入图工作流和动态工作流，新项目应评估对应机制；这里的版本运行通过不代表模板API是所有新项目的首选。两种框架的运行结果不能直接等同于真实模型效果。
扩展边界
主流程不包含切分器、向量库或重排序器。MCP示例使用本地stdio，不包含HTTP认证、远程权限控制或服务可用性测试。主流程遇到资料不足直接停止，若需要多次改写查询，应显式增加状态、次数上限和验收案例。