LLM之Agent(二十九)|LangChain 1.0核心组件介绍

大模型中间件机器学习
 智能体是将LLM与工具相结合,创建能够对任务进行推理、决定使用哪些工具并迭代地寻求解决方案的系统。


 LangChain 1.0把创建智能体统一为create\_agent接口。AI Agent循环调用各种工具运行,直到满足停止条件为止——即模型返回最终输出或达到迭代限制次数。

picture.image

create_agent 的底层是 LangGraph,是一种图结构。

下面介绍一下LangChain 1.0的核心组件:

一、Model

 模型是智能体的推理引擎,LangChain 1.0支持静态和动态模型两种方式。

1.1 静态模型

 静态模型就是在创建代理时,通过字符串配置一次,并在整个执行过程中保持不变。这是最常见、最直接的方法。

下面,我们看一下如何指定静态模型:

  
from langchain.agents import create_agent  
  
agent = create_agent(  
    "gpt-5",  
    tools=tools  
)
 当然,为了更好地控制模型,我们可以配置更多参数,比如ChatOpenAI接口,可以配置的参数如下所示:
  
from langchain.agents import create_agent  
from langchain_openai import ChatOpenAI  
  
model = ChatOpenAI(  
    model="gpt-5",  
    temperature=0.1,  
    max_tokens=1000,  
    timeout=30  
    # ... (other params)  
)  
agent = create_agent(model, tools=tools)

1.2 动态模型

 动态模型是指在Agent运行时,动态选择大模型,在LangChain 1.0是通过@wrap\_model\_call装饰器创建中间件来实现动态选择模型的:
  
from langchain_openai import ChatOpenAI  
from langchain.agents import create_agent  
from langchain.agents.middleware import wrap_model_call, ModelRequest, ModelResponse  
  
  
basic_model = ChatOpenAI(model="gpt-4o-mini")  
advanced_model = ChatOpenAI(model="gpt-4o")  
  
@wrap_model_call  
def dynamic_model_selection(request: ModelRequest, handler) -> ModelResponse:  
    """Choose model based on conversation complexity."""  
    message_count = len(request.state["messages"])  
  
    if message_count > 10:  
        # Use an advanced model for longer conversations  
        model = advanced_model  
    else:  
        model = basic_model  
  
    request.model = model  
    return handler(request)  
  
agent = create_agent(  
    model=basic_model,  # Default model  
    tools=tools,  
    middleware=[dynamic_model_selection]  
)

使用结构化输出时,不支持预绑定模型(已调用 bind_tools 的模型)。如果需要使用结构化输出进行动态模型选择,请确保传递给中间件的模型未预先绑定。

二、Tool

 工具可以赋能智能体更强的执行力,智能体除了可以简单的调用绑定工具的LLM,还能实现以下功能:
  • 连续调用多个工具(由单个提示触发)

  • 在适当的时候并行调用工具

  • 基于先前的结果动态工具选择

  • 工具重试逻辑和错误处理

  • 跨工具调用保持状态持久性

2.1 定义工具

 首先,使用@tool装饰器定义各种工具,将工具列表传递给Agent,如下所示:
  
from langchain.tools import tool  
from langchain.agents import create_agent  
  
  
@tool  
def search(query: str) -> str:  
    """Search for information."""  
    return f"Results for: {query}"  
  
@tool  
def get_weather(location: str) -> str:  
    """Get weather information for a location."""  
    return f"Weather in {location}: Sunny, 72°F"  
  
agent = create_agent(model, tools=[search, get_weather])

如果提供的工具列表为空,则Agent将由一个没有工具调用功能的 LLM 节点组成。

2.2 工具错误处理

 处理工具错误,可以使用 @wrap\_tool\_call装饰器 自定义工具错误中间件:
  
from langchain.agents import create_agent  
from langchain.agents.middleware import wrap_tool_call  
from langchain_core.messages import ToolMessage  
  
  
@wrap_tool_call  
def handle_tool_errors(request, handler):  
    """Handle tool execution errors with custom messages."""  
    try:  
        return handler(request)  
    except Exception as e:  
        # Return a custom error message to the model  
        return ToolMessage(  
            content=f"Tool error: Please check your input and try again. ({str(e)})",  
            tool_call_id=request.tool_call["id"]  
        )  
  
agent = create_agent(  
    model="gpt-4o",  
    tools=[search, get_weather],  
    middleware=[handle_tool_errors]  
)
当工具执行失败时,Agent将返回一个包含自定义错误消息的 ToolMessage
  
[      ...      ToolMessage(          content="Tool error: Please check your input and try again. (division by zero)",          tool_call_id="..."      ),      ...  ]

2.3 ReAct 循环中的工具使用

 智能体遵循 ReAct(“推理 + 行动”)模式,在简短的推理步骤和有针对性的工具调用之间交替进行,并将由此产生的观察结果输入到后续决策中,直到能够给出最终答案。

三、系统提示词

常见的系统提示词是静态的,指定方式如下所示:

  
agent = create_agent(  
    model,  
    tools,  
    system_prompt="You are a helpful assistant. Be concise and accurate."  
)

如果没有提供 system_prompt ,代理将直接从消息中推断其任务

3.1 动态系统提示词

 如果在运行时需要动态修改系统提示词,LangChain 1.0也提供了中间件实现方式,是通过@dynamic\_prompt 装饰器实现的:
  
from typing import TypedDict  
  
from langchain.agents import create_agent  
from langchain.agents.middleware import dynamic_prompt, ModelRequest  
  
  
class Context(TypedDict):  
    user_role: str  
  
@dynamic_prompt  
def user_role_prompt(request: ModelRequest) -> str:  
    """Generate system prompt based on user role."""  
    user_role = request.runtime.context.get("user_role", "user")  
    base_prompt = "You are a helpful assistant."  
  
    if user_role == "expert":  
        return f"{base_prompt} Provide detailed technical responses."  
    elif user_role == "beginner":  
        return f"{base_prompt} Explain concepts simply and avoid jargon."  
  
    return base_prompt  
  
agent = create_agent(  
    model="gpt-4o",  
    tools=[web_search],  
    middleware=[user_role_prompt],  
    context_schema=Context  
)  
  
# The system prompt will be set dynamically based on context  
result = agent.invoke(  
    {"messages": [{"role": "user", "content": "Explain machine learning"}]},  
    context={"user_role": "expert"}  
)

四、结构化输出

 在某些情况下,希望代理以特定格式返回输出,LangChain 1.0通过 response\_format 参数提供了结构化输出的策略。

4.1 ToolStrategy

 ToolStrategy 使用人工定义好的结构来生成结构化输出,适用于任何支持工具调用的模型:
  
from pydantic import BaseModel  
from langchain.agents import create_agent  
from langchain.agents.structured_output import ToolStrategy  
  
  
class ContactInfo(BaseModel):  
    name: str  
    email: str  
    phone: str  
  
agent = create_agent(  
    model="gpt-4o-mini",  
    tools=[search_tool],  
    response_format=ToolStrategy(ContactInfo)  
)  
  
result = agent.invoke({  
    "messages": [{"role": "user", "content": "Extract contact info from: John Doe, john@example.com, (555) 123-4567"}]  
})  
  
result["structured_response"]  
# ContactInfo(name='John Doe', email='john@example.com', phone='(555) 123-4567')

4.2 ProviderStrategy

 ProviderStrategy 使用模型提供商的原生结构化输出生成功能。这种方式更可靠,但仅适用于支持原生结构化输出的提供商(例如 OpenAI):
  
from langchain.agents.structured_output import ProviderStrategy  
  
agent = create_agent(  
    model="gpt-4o",  
    response_format=ProviderStrategy(ContactInfo)  
)

从 langchain 1.0 开始,不再支持直接传递 schema(例如, response_format=ContactInfo )。必须显式使用 ToolStrategy 或 ProviderStrategy

五、Memory

  长期记忆这里不谈,存储在state中的信息可以被视为智能体的短期记忆 :


 我们可以创建custom state模式,它必须扩展 AgentState 作为 TypedDict 。

定义custom state有两种方法:

  • 通过中间件 (首选)
  • 通过 create_agent 上的 state_schema

比起在 create_agent 中通过 state_schema 定义自定义状态,通过中间件定义自定义状态是更可取的,因为它允许你将状态扩展在概念上限制在相关的中间件和工具范围内。

为了向后兼容, create_agent 仍然支持 state_schema 。

5.1 通过中间件定义状态

 当custom state需要被附加到该中间件的特定中间件钩子和工具访问时,可以使用中间件来定义自定义状态。
  
from langchain.agents import AgentState  
from langchain.agents.middleware import AgentMiddleware  
from typing import Any  
  
  
class CustomState(AgentState):  
    user_preferences: dict  
  
class CustomMiddleware(AgentMiddleware):  
    state_schema = CustomState  
    tools = [tool1, tool2]  
  
    def before_model(self, state: CustomState, runtime) -> dict[str, Any] | None:  
        ...  
  
agent = create_agent(  
    model,  
    tools=tools,  
    middleware=[CustomMiddleware()]  
)  
  
# The agent can now track additional state beyond messages  
result = agent.invoke({  
    "messages": [{"role": "user", "content": "I prefer technical explanations"}],  
    "user_preferences": {"style": "technical", "verbosity": "detailed"},  
})

5.2 通过 state_schema 定义状态

 使用 state\_schema 参数作为快捷方式来定义仅在工具中使用的自定义状态。
  
from langchain.agents import AgentState  
  
  
class CustomState(AgentState):  
    user_preferences: dict  
  
agent = create_agent(  
    model,  
    tools=[tool1, tool2],  
    state_schema=CustomState  
)  
# The agent can now track additional state beyond messages  
result = agent.invoke({  
    "messages": [{"role": "user", "content": "I prefer technical explanations"}],  
    "user_preferences": {"style": "technical", "verbosity": "detailed"},  
})

从 langchain 1.0 开始,自定义状态模式必须是 TypedDict 类型。不再支持 Pydantic 模型和数据类

六、Streaming

 我们已经了解了如何使用 invoke 调用代理来获取最终响应。如果代理执行多个步骤,这可能需要一些时间。为了显示中间进度,我们可以实时发送消息流。
  
for chunk in agent.stream({  
    "messages": [{"role": "user", "content": "Search for AI news and summarize the findings"}]  
}, stream_mode="values"):  
    # Each chunk contains the full state at that point  
    latest_message = chunk["messages"][-1]  
    if latest_message.content:  
        print(f"Agent: {latest_message.content}")  
    elif latest_message.tool_calls:  
        print(f"Calling tools: {[tc['name'] for tc in latest_message.tool_calls]}")

七、Middleware

 中间件提供了强大的扩展性,可用于在执行的不同阶段自定义代理行为。您可以使用中间件来:
  • 在调用模型之前处理状态(例如,消息修剪、上下文注入)
  • 修改或验证模型的响应(例如,防护措施、内容过滤)
  • 使用自定义逻辑处理工具执行错误
  • 基于状态或上下文实现动态模型选择
  • 添加自定义日志记录、监控或分析功能
  • 中间件可以无缝集成到代理的执行图中,使您能够在关键点拦截和修改数据流,而无需更改核心代理逻辑。

中间件有很多装饰器,具体如下所示:

picture.image

0
0
0
0
关于作者
关于作者

文章

0

获赞

0

收藏

0

相关资源
字节跳动 XR 技术的探索与实践
火山引擎开发者社区技术大讲堂第二期邀请到了火山引擎 XR 技术负责人和火山引擎创作 CV 技术负责人,为大家分享字节跳动积累的前沿视觉技术及内外部的应用实践,揭秘现代炫酷的视觉效果背后的技术实现。
相关产品
评论
未登录
看完啦,登录分享一下感受吧~
暂无评论