Langchain(下)-编程阁

Agent部分

C-898S-989D-82N快点岛币吧

组件

Tool

的问题和工具交给大语言模型然后大语言模型进行1个意图识别决定是否要执行工具如果需要执行工具的话，执行工具并返回结果再返回给大语言模型然后代言模型根据结果生成我们的答案

SDK

这里直接调用openai的sdk

结果是

ChatCompletionMessage(content='', refusal=None, role='assistant', annotations=None, audio=None, function_call=None, tool_calls=[ChatCompletionMessageFunctionToolCall(id='call_579422a0074c40f28d68f1', function=Function(arguments='{"city": "北京", "date": "2026-05-22"}', name='get_weather'), type='function', index=0)])

from openai import OpenAI import json client = OpenAI() # 1. 通过JSON结构定义工具，包括工具名称，描述，参数等 tools = [ { "type": "function", "function": { "name": "get_weather", "description": "Get today's weather for a location.", "parameters": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称, e.g. San Francisco", }, "date" :{ "type":"string", "description":"想要查询的天气的日期, e.g. 2023-12-25" } }, "required": ["city","date"], "additionalProperties": False, }, "strict": True, }, }, ] def get_weather(city,date): return f"{city} on {date} is cloudy with a chance of rain." messages = [ {"role": "user", "content": "What is the weather like in 北京 on 2024-12-25?"} ] # 2. Prompt the model with tools defined response = client.chat.completions.create( model="gpt-4.1", messages=messages, tools=tools, ) messages.append(response.choices[0].message) for tool_call in response.choices[0].message.tool_calls or []: if tool_call.function.name == "get_weather": # 3. 执行工具函数的逻辑 args = json.loads(tool_call.function.arguments) weather = get_weather(args["city"],args["date"]) # 4. 将工具函数的执行结果添加到消息列表中 messages.append( { "role": "tool", "tool_call_id": tool_call.id, "content": json.dumps({"weather": weather}), } ) response = client.chat.completions.create( model="gpt-4.1", messages=messages, tools=tools, ) # 5. 模型会根据工具函数的执行结果，生成最终的回复 print(response.choices[0].message.content)

LANGCHAIN方式

import json import os from dotenv import load_dotenv from langchain_core.messages import HumanMessage, ToolMessage from langchain_core.tools import tool from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field # 1 使用langchain装饰器定义工具方法和约束 class GetWeatherArgs(BaseModel): # 定义参数city：字符串类型，带描述（大模型会根据这个描述理解参数含义） city: str = Field(description="城市名称, 比方说 北京市， 武汉市，需要传递城市的全名") # 定义参数date：字符串类型，指定日期格式 date: str = Field(description="天气日期，格式是yyyy-MM-dd") @tool(description="获取城市在特定日期的天气",args_schema=GetWeatherArgs) def get_weather(city: str, date: str): # 模拟天气查询，实际项目中这里可以调用真实天气API return f'城市{city}在{date}的天气是晴朗的' # 2 创建llm load_dotenv() # 初始化LLM大模型：使用ChatOpenAI调用 llm = ChatOpenAI( model="qwen-turbo", api_key=os.getenv("OPENAI_API_KEY"), # 从.env读取千问的API Key base_url="https://dashscope.aliyuncs.com/compatible-mode/v1", # 千问的OpenAI兼容接口地址 ) # 3 把工具和llm绑定 llm_with_tools = llm.bind_tools([get_weather]) # 4 构建message human_message = HumanMessage(content="北京在2026-05-22的天气是多少？") # 5 第一次调用llm：意图识别 res = llm_with_tools.invoke([human_message]) ai_message = res print("=== 第一次调用（大模型决定调用工具）===") print(res) # 6 从第一次llm返回结果获取工具名称和参数，进行工具调用 tool_call = res.tool_calls[0]['args'] # tool_res调用工具之后的结果 tool_res = get_weather.invoke(tool_call) # 7 构建message：包含问题 + 第一次调用llm结果 + 工具执行结果 # 8 第二次调用llm tool_id = res.tool_calls[0]['id'] tool_message = ToolMessage(tool_call_id=tool_id, content=tool_res) # human_message: 用户问题 # ai_message：第一次调用llm结果 # tool_message：工具执行结果 final_result = llm_with_tools.invoke([human_message, ai_message, tool_message]) # 打印最终回答 print('\n=== 模型第二次的回复结果（最终答案）===\n', final_result)

createagent

import os from dotenv import load_dotenv from langchain.agents import create_agent from langchain_core.messages.tool import tool_call from langchain_core.tools import tool from langchain_openai import ChatOpenAI # 创建llm load_dotenv() llm = ChatOpenAI( model="qwen-turbo", # 千问模型 api_key=os.getenv("OPENAI_API_KEY"), base_url="https://dashscope.aliyuncs.com/compatible-mode/v1" ) # 创建工具方法 @tool def get_weather(city:str,date:str): """ 获取城市在特定日期的天气 """ return f'在{city} {date}的天气是晴朗的' # 创建agent agent = create_agent( model=llm, tools=[get_weather], ) # 调用agent，获取结果 res = agent.invoke({"messages":[{"role":"user", "content":"北京在2026-05-22的天气是多少？"}]}) print(res)

MCP

MCP的工作流程如下所示：（1）握手与能力声明（Handshake & Discovery）：当AI应用程序（Host）启动时，它会根据配置启动 MCP Server（通常是一个子进程）。Host 记住了这些工具的名字和用法说明，但此时并没有执行任何代码。（2）用户提问与上下文注入（Context Injection）：User提出问题后，Host 把用户的这个问题，加上刚才 Server 汇报的工具说明书，一起打包发给 LLM。（3）模型决策（LLM Reasoning）：LLM决定是否需要去调用相关工具。（4）路由与执行（Routing & Execution）：Host 收到指令，发现是要调用 MCP 工具，Host 通过 MCP 协议（JSON-RPC）给 MCP Server 发消息，Server 收到请求，在它自己的进程（如果是StreamableHttp方式，则是在远端执行）里执行相关操作。所有的驱动依赖、复杂逻辑都在 Server 端。（5）结果回传（Result Feedback）：Server 拿到结果，打包成 MCP 响应发回给 Host，Host再将相关结果，传递给大模型，让大模型做下一步的输出。

服务端

from mcp.server import FastMCP ## stdio传输协议， # 特点：本身不是服务器方式，使用本地命令行调用， # 不需要启动服务器端，在客户端启动和调用 # 定义FastMCP实例 mcp = FastMCP() # 创建工具、资源和提示模板 @mcp.tool() def get_weather(city:str,date:str) -> str: return f"在{city}，{date}的天气是下雨的" @mcp.resource("greeting://default") def get_greeting() -> str: return "Hello from static resource!" @mcp.prompt() def greet_user(name: str, style: str = "friendly") -> str: styles = { "friendly": "写一句友善的问候", "formal": "写一句正式的问候", "casual": "写一句轻松的问候", } return f"为{name}{styles.get(style, styles['friendly'])}" if __name__ == "__main__": mcp.run(transport="stdio")

客户端

from mcp.client.streamable_http import streamable_http_client from mcp import ClientSession async def main(): url = "http://localhost:8000/mcp" async with streamable_http_client(url=url) as (read_stream,write_stream,_): async with ClientSession(read_stream=read_stream,write_stream=write_stream) as session: # 1、初始化 await session.initialize() # 2、获取到server端所有工具 tools = await session.list_tools() print(tools) # 3、调用tool result = await session.call_tool(name="get_temperature",arguments={"city":"北京","date":"2023-12-12"}) print(result) if __name__ == "__main__": import asyncio asyncio.run(main())