16. LangChain自主智能体（Autonomous Agent）：模拟人类工作流的进阶设计

引言：当AI学会"思考"与"行动"

2025年某跨国律所的合同审查智能体，通过自主规划任务流，将平均处理时间从8小时缩短至23分钟。本文将基于LangChain的AgentExecutor与Deepseek-R1，揭示如何构建能自主决策、动态调整的智能体系统。

一、自主智能体核心架构

1.1 智能体能力层级

1.2 LangChain 关键组件

组件	功能描述	生产环境要求
AgentExecutor	任务流程控制中枢	支持断点续执行
MemoryGraph	跨会话状态保持	支持记忆快照

二、四步构建合同审查智能体

2.1 工具注册与权限控制

from langchain_core.tools import Tool
from langchain_community.agent_toolkits import SQLDatabaseToolkit
from langchain_community.utilities import SQLDatabase
from langchain_ollama import ChatOllama

# 法律专用工具
def search_legal_database(clause_id: str) -> str:return f"法律条款 {clause_id} 内容：..."

# 权限控制装饰器
def rbac_check(user_role: str, allowed_roles: list):def decorator(func):def wrapper(*args, **kwargs):if user_role not in allowed_roles:raise ValueError(f"角色 {user_role} 无权限执行此操作")return func(*args, **kwargs)return wrapperreturn decorator

user_role = "junior_lawyer"  # 实际应从用户会话获取

llm = ChatOllama(model="deepseek-r1")

# 正确初始化SQL工具包
toolkit = SQLDatabaseToolkit(db=SQLDatabase.from_uri("数据库连接信息"),llm=llm
)

# 带权限的工具
contract_tools = [Tool(name="clause_lookup",func=rbac_check(user_role, allowed_roles=["junior_lawyer", "senior_lawyer"])(search_legal_database),description="根据条款编号查询法律条文"),*toolkit.get_tools()
]

2.2 动态记忆管理

from langchain.memory import ConversationBufferWindowMemory
import json

class RollbackMemory:def __init__(self, file_path):self.file_path = file_pathself.memory = ConversationBufferWindowMemory(memory_key="chat_history",k=5,  # 保留最近5轮对话return_messages=True)
def save_state(self, session_id):state = {"history": self.memory.load_memory_variables({}),"session_id": session_id}with open(f"{self.file_path}/{session_id}.json", "w") as f:json.dump(state, f)
def rollback(self, session_id, steps=3):with open(f"{self.file_path}/{session_id}.json") as f:state = json.load(f)# 实现回滚逻辑（如截断历史）return state["history"][:-steps]

2.3 自监控与优化

from langchain.callbacks import FileCallbackHandler

class AlertCallback(FileCallbackHandler):def on_tool_start(self, serialized, input_str, **kwargs):tool_name = serialized["name"]if self._check_abuse(tool_name):raise ValueError(f"工具 {tool_name} 调用过于频繁")
def _check_abuse(self, tool_name):# 实现调用频率检查return False

2.4 执行

from langchain.agents import initialize_agent, AgentType

# 记忆管理
memory = RollbackMemory("./agent_states")

# 必须指定日志文件路径
alert_cb = AlertCallback("alerts.log")
agent_executor = initialize_agent(tools=contract_tools,llm=llm,agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,  # 使用预设模板memory=memory.memory,verbose=True,handle_parsing_errors=True,max_iterations=5,callbacks=[alert_cb]
)

result = agent_executor.invoke({"input": "审查本合同第3.2条款"})
print(result)

输出为：

{'input': '审查本合同第3.2条款', 'chat_history': [], 'output': '<think>\n好的，我现在需要处理用户关于审查合同第三条第二款的需求。首先，我应该明确用户的主要目标是什么。他们希望检查这则法律条款的内容是否符合规定、有没有遗漏或错误。\n\n接下来，我会考虑有哪些工具可以使用来完成这个任务。根据提供的工具列表，有五个可用的工具：clause_lookup、sql_db_query、sql_db_schema、sql_db_list_tables 和 sql_db_query_checker。这些工具分别用于查找条款、执行SQL查询、获取表结构、列出所有表格以及检查查询的有效性。\n\n因为用户现在需要审查的是第3.2条款，所以我首先应该使用clause_lookup工具来获取该条款的具体内容和相关法律条文。这样可以帮助我了解条款的详细信息，确保我没有遗漏任何关键点或者误读条款的内容。\n\n在使用clause_lookup之前，可能还需要确认是否有相关的数据库支持，因为有时候条款可能会存储在一个特定的数据库表中。因此，我会先调用sql_db_list_tables工具来列出所有存在的表格名称，以便确定哪个表格包含第3.2条款。假设现在我运行了这个工具，并得到了结果，发现对应的表格名称是contract_clauses_table。\n\n接下来，我会将该条款请求插入到contract_clauses_table中。使用clause_lookup工具查询条款内容后，会得到一个包含条款详细信息的响应。然后，我需要将这些条款内容与现有的法律条文进行比较，确保它们符合相关法律法规，并且没有冲突的地方。\n\n在获取了条款的具体内容之后，可能会发现一些需要注意的地方，比如条款中的术语是否准确，是否有遗漏的部分，或者是否有与其他条款之间的关系需要考虑。这时，我会利用sql_db_query工具编写一个查询语句，检索该条款的相关数据，并检查是否存在与之相关的其他条款或条件。\n\n如果在执行SQL查询时遇到了问题，比如未知的列名或其他错误，我将使用sql_db_schema工具来获取该表格的详细结构信息。这可以帮助我理解查询中出现的问题所在，从而调整和修正我的SQL查询语句，确保能够正确地提取到所需的条款内容。\n\n此外，在处理过程中，如果遇到不清晰或模糊的部分，比如条款中的某些术语或者条件，我会再次使用clause_lookup工具来查找相关的法律条文，确保对条款的理解是准确的。如果有疑问或者不确定的地方，我可能会参考官方的法律解释文档或者咨询相关专家，以确保审查的全面性和准确性。\n\n最后，在完成所有检查之后，如果一切都符合预期，那么我就可以确定第3.2条款是符合规定的，并且审查结果没有问题。这样用户就可以放心地使用该条款了。\n</think>\n\n{\n  "action": "clause_lookup",\n  "action_input": "contract_clauses_table, clause 3.2"\n}\n'}

三、企业级案例：跨境合同审核

3.1 智能体工作流

3.2 性能指标对比

指标	人工处理	智能体处理
平均耗时	8.2h	23min
条款覆盖率	78%	96%
风险漏检率	12%	3%
版本迭代周期	2周	实时更新

四、避坑指南：自主系统七大陷阱

无限递归

# 解决方案：强制深度限制
from langchain.agents import AgentExecutor

agent = create_your_agent()  # 你的Agent初始化代码
agent_executor = AgentExecutor(agent=agent,tools=tools,max_iterations=10,  # 关键参数：强制限制最大步数max_execution_time=30, # 代理从开始执行到强制终止的最长时间（单位：秒）early_stopping_method="generate",  # 达到限制时自动终止handle_parsing_errors=True  # 避免解析错误导致循环
)