LangChain v1 入门：用 create_agent 跑通第一个 Agent（无需 API Key）

"""Print the interpreter and package versions the other scripts in this lab were recorded with."""
import importlib.metadata as metadata
import sys

print("python", sys.version.split()[0])
for package in ["langchain", "langchain-core", "langgraph", "langchain-openai", "openai", "pydantic"]:
    print(package, metadata.version(package))

python 3.12.12
langchain 1.4.7
langchain-core 1.7.0
langgraph 1.2.17
langchain-openai 1.7.2
openai 3.37.0
pydantic 2.14.1

"""The smallest useful agent: one tool, one system prompt, one invoke call."""
from langchain.agents import create_agent

from shared import RecordingModel, get_weather, weather_script

model = RecordingModel(script=weather_script())  # stands in for a real chat model
agent = create_agent(model, tools=[get_weather], system_prompt="你是天气助手，用中文简短回答。")

result = agent.invoke({"messages": [{"role": "user", "content": "北京天气怎么样？"}]})

print("result type:", type(result).__name__)
print("result keys:", list(result))
for message in result["messages"]:
    if message.type == "ai" and message.tool_calls:
        call = message.tool_calls[0]
        print(f"ai: 调用工具 {call['name']}({call['args']})")
    else:
        print(f"{message.type}: {message.content}")
print("最终回答:", result["messages"][-1].content)

result type: dict
result keys: ['messages']
human: 北京天气怎么样？
ai: 调用工具 get_weather({'city': '北京'})
tool: 晴，22°C
ai: 北京现在晴，22°C。
最终回答: 北京现在晴，22°C。

"""What the model actually receives: the tool definition and the messages of every round."""
import json

from langchain.agents import create_agent
from langchain_core.utils.function_calling import convert_to_openai_tool

from shared import RecordingModel, get_weather, weather_script

print("-- 模型收到的工具定义 --")
print(json.dumps(convert_to_openai_tool(get_weather), ensure_ascii=False, indent=2))

model = RecordingModel(script=weather_script())
agent = create_agent(model, tools=[get_weather], system_prompt="你是天气助手，用中文简短回答。")
agent.invoke({"messages": [{"role": "user", "content": "北京天气怎么样？"}]})


def show(message):
    if message.type == "ai" and message.tool_calls:
        return f"ai(tool_calls={[(c['name'], c['args']) for c in message.tool_calls]})"
    return f"{message.type}({message.content})"


print("-- 绑定到模型的工具 --")
print(model.bound)
for number, received in enumerate(model.inputs, start=1):
    print(f"-- 第 {number} 轮，模型收到 {len(received)} 条消息 --")
    for message in received:
        print(" ", show(message))

-- 模型收到的工具定义 --
{
  "type": "function",
  "function": {
    "name": "get_weather",
    "description": "查询城市当前天气。",
    "parameters": {
      "properties": {
        "city": {
          "type": "string"
        }
      },
      "required": [
        "city"
      ],
      "type": "object"
    }
  }
}
-- 绑定到模型的工具 --
['get_weather']
-- 第 1 轮，模型收到 2 条消息 --
  system(你是天气助手，用中文简短回答。)
  human(北京天气怎么样？)
-- 第 2 轮，模型收到 4 条消息 --
  system(你是天气助手，用中文简短回答。)
  human(北京天气怎么样？)
  ai(tool_calls=[('get_weather', {'city': '北京'})])
  tool(晴，22°C)

"""Which shapes of invoke() input does create_agent accept, and what happens to the question."""
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage

from shared import RecordingModel, get_weather, weather_script

QUESTION = "北京天气怎么样？"
CASES = [
    ("messages 是 dict 列表", {"messages": [{"role": "user", "content": QUESTION}]}),
    ("messages 是元组列表", {"messages": [("user", QUESTION)]}),
    ("messages 是 HumanMessage 列表", {"messages": [HumanMessage(QUESTION)]}),
    ("messages 直接是字符串", {"messages": QUESTION}),
    ("整个输入是字符串", QUESTION),
    ('旧写法 {"input": ...}', {"input": QUESTION}),
    ('空列表 {"messages": []}', {"messages": []}),
]

for label, payload in CASES:
    model = RecordingModel(script=weather_script())
    agent = create_agent(model, tools=[get_weather], system_prompt="你是天气助手，用中文简短回答。")
    try:
        agent.invoke(payload)
    except Exception as error:  # report the exception type and its first line
        first_line = (str(error).splitlines() or ["(无说明文字)"])[0]
        print(f"{label}: 报错 {type(error).__name__}: {first_line}")
        continue
    first_round = [f"{m.type}" for m in model.inputs[0]]
    sent = QUESTION in "".join(str(m.content) for m in model.inputs[0])
    print(f"{label}: 不报错；第 1 轮模型收到 {first_round}；问题送达模型: {sent}")

messages 是 dict 列表: 不报错；第 1 轮模型收到 ['system', 'human']；问题送达模型: True
messages 是元组列表: 不报错；第 1 轮模型收到 ['system', 'human']；问题送达模型: True
messages 是 HumanMessage 列表: 不报错；第 1 轮模型收到 ['system', 'human']；问题送达模型: True
messages 直接是字符串: 不报错；第 1 轮模型收到 ['system', 'human']；问题送达模型: True
整个输入是字符串: 报错 InvalidUpdateError: Expected dict, got 北京天气怎么样？
旧写法 {"input": ...}: 不报错；第 1 轮模型收到 ['system']；问题送达模型: False
空列表 {"messages": []}: 不报错；第 1 轮模型收到 ['system']；问题送达模型: False

"""Beginner mistakes around create_agent, each reproduced with its exact outcome."""
from langchain.agents import create_agent
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage
from langchain_core.outputs import ChatGeneration, ChatResult

from shared import RecordingModel, get_weather, weather_script

INPUT = {"messages": [{"role": "user", "content": "北京天气怎么样？"}]}


def first_line(error: Exception) -> str:
    return (str(error).splitlines() or ["(无说明文字)"])[0]


def report(label: str, action) -> None:
    print(label)
    try:
        outcome = action()
    except Exception as error:
        print(f"    -> {type(error).__name__}: {first_line(error)}")
    else:
        print(f"    -> 成功: {outcome}")


def agent_with(tools):
    return create_agent(RecordingModel(script=weather_script()), tools=tools)


class NoToolCallingModel(BaseChatModel):
    """A chat model that never implemented bind_tools()."""

    @property
    def _llm_type(self) -> str:
        return "no-tool-calling"

    def _generate(self, messages, stop=None, run_manager=None, **kwargs) -> ChatResult:
        return ChatResult(generations=[ChatGeneration(message=AIMessage(content="你好"))])


def plain_function(city: str) -> str:
    """查询城市当前天气。"""
    return "晴"


def without_docstring(city: str) -> str:
    return "晴"


def read_result_as_message():
    result = agent_with([get_weather]).invoke(INPUT)
    return result.content


def read_result_correctly():
    result = agent_with([get_weather]).invoke(INPUT)
    return result["messages"][-1].content


def model_without_tool_calling():
    agent = create_agent(NoToolCallingModel(), tools=[get_weather])
    print("    (create_agent 返回了，没有报错)")
    return agent.invoke(INPUT)["messages"][-1].content


report("1. 把 invoke 的返回值当消息读：result.content", read_result_as_message)
report('2. 正确读法：result["messages"][-1].content', read_result_correctly)
report("3. tools 传了单个工具而不是列表：tools=get_weather", lambda: agent_with(get_weather))
report("4. 不用 @tool，直接传带类型标注和 docstring 的普通函数", lambda: agent_with([plain_function]).invoke(INPUT)["messages"][-1].content)
report("5. 普通函数没有 docstring", lambda: agent_with([without_docstring]))
report("6. 模型没有实现 bind_tools", model_without_tool_calling)
report('7. 用字符串指定模型但没有设置密钥：create_agent("openai:gpt-5.5", ...)', lambda: create_agent("openai:gpt-5.5", tools=[get_weather]))

1. 把 invoke 的返回值当消息读：result.content
    -> AttributeError: 'dict' object has no attribute 'content'
2. 正确读法：result["messages"][-1].content
    -> 成功: 北京现在晴，22°C。
3. tools 传了单个工具而不是列表：tools=get_weather
    -> ValueError: The first argument must be a string or a callable with a __name__ for tool decorator. Got <class 'tuple'>
4. 不用 @tool，直接传带类型标注和 docstring 的普通函数
    -> 成功: 北京现在晴，22°C。
5. 普通函数没有 docstring
    -> ValueError: Function must have a docstring if description not provided.
6. 模型没有实现 bind_tools
    (create_agent 返回了，没有报错)
    -> NotImplementedError: (无说明文字)
7. 用字符串指定模型但没有设置密钥：create_agent("openai:gpt-5.5", ...)
    -> OpenAIError: Missing credentials. Please pass an `api_key`, `workload_identity`, `admin_api_key`, or set the `OPENAI_API_KEY` or `OPENAI_ADMIN_KEY` environment variable.

from langchain.agents import create_agent

agent = create_agent(
    model="openai:gpt-5.5",  # 官方文档示例中的模型字符串；可用的模型名以官方文档为准
    tools=[get_weather],
    system_prompt="你是天气助手，用中文简短回答。",
)