简单创建智能体并调用

"""
可参考官方文档地址：https://docs.langchain.com/oss/python/langchain/agents
智能体会遵循 ReAct（“推理+行动”）模式，交替进行简短的推理步骤和针对性工具调用，并将所得观察反馈到后续决策中，直到能够给出最终答案。

第一步：初始化测试用的工具
"""
from langchain_core.tools import tool
from pydantic import BaseModel, Field


class AddInputArgs(BaseModel):
    a: float = Field(description="第一个数字", example=1)
    b: float = Field(description="第二个数字", example=2)


@tool(
    description="对输入的两个数字执行加法运算",
    args_schema=AddInputArgs,
    return_direct=False
)
def add(a, b) -> float:
    return a + b


"""
第二步：初始化大模型
这里直接初始化一个千问大模型
"""
import os
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

llm = ChatOpenAI(
    model="qwen-max-latest",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    streaming=True,
)

"""
第三步：创建智能体
通过 langchian.agent的create_agent方法
"""
from langchain.agents import create_agent

agent = create_agent(
    model=llm,
    tools=[add],
    system_prompt="你是一个擅长计算的专家，你会使用数学工具完成计算"
)

"""
第四步：调用智能体
"""
result = agent.invoke(input={"messages": [{"role": "user", "content": "请计算1+2的值"}]})
"""美化 打印result的消息列表"""
for msg in result['messages']:
    t = type(msg).__name__
    if t == 'HumanMessage':
        print(f"👤 {msg.content}")
    elif t == 'AIMessage':
        if hasattr(msg, 'tool_calls') and msg.tool_calls:
            for tool in msg.tool_calls:
                args = ', '.join(f'{k}={v}' for k, v in tool['args'].items())
                print(f"🤖 调用 {tool['name']}({args})")
        elif msg.content:
            print(f"🤖 {msg.content}")
    elif t == 'ToolMessage':
        print(f"🔧 {msg.name} → {msg.content}")

控制台输出内容：
👤 请计算1+2的值
🤖 调用 add(a=1, b=2)
🔧 add → 3.0
🤖 计算结果为 \(1 + 2 = 3\)。

使用Saver保存会话消息

langchian的对话消息，有多种保存方式，可以使用redis、MongoDB、或者写一个自定义saver保存到本地都是可以的

• 在早期可以使用社区提供的ChatMessageHistory 来实现简单的会话消息保存，保存和提取的动作都需要自己维护
  后期官方提供了MemorySaver 这是完整的会话管理系统，langchian会自动处理一起

案例代码

"""
agent会话保存
可以使用临时内存保存、也可以使用redis、MongoDB甚至是自己实现本地文件保存

"""
import asyncio
import json
import os
from dotenv import load_dotenv
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langchain_core.runnables import RunnableConfig
from langchain.agents import create_agent

load_dotenv()

"""
第一步：定义Saver
"""
from langgraph.checkpoint.memory import MemorySaver

memory_saver = MemorySaver()

"""
第二步：初始化大模型
"""
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="qwen-max-latest",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    streaming=True,
)

"""
第三步：创建react Agent
并传入checkpointer
"""
agent = create_agent(
    model=llm,
    checkpointer=memory_saver,
    debug=True,
    system_prompt=SystemMessage(content="你是3DM的游戏助手，专注游戏问题"),
)


"""
第四步 创建RunnableConfig,用来传入ThreadId 也就是会话标识
"""
config = RunnableConfig(configurable={"thread_id": 1}, recursion_limit=100)


"""
第五步 流式调用agent连续对话
"""


async def run_conversation():
    print("=" * 50)
    print("游戏助手已启动！输入 'exit' 退出，输入 'history' 查看历史记录")
    print("=" * 50)

    conversation_history = []  # 本地存储对话历史

    while True:
        user_input = input("\n用户：")

        # 退出命令
        if user_input.lower() == "exit":
            print("\n对话结束！")
            break

        print("\n助手：", end="", flush=True)
        try:
            # 流式异步调用大模型
            async for chunk in agent.astream(input={"messages": user_input}, config=config):
                # 根据你提供的格式解析chunk
                await process_chunk(chunk)
        except Exception as e:
            print(f"\n[错误] 调用大模型时出错: {e}")
            import traceback
            traceback.print_exc()
        print()  # 换行



async def process_chunk(chunk):
    """处理流式返回的chunk"""
    if isinstance(chunk, dict):
        # 处理 [values] 类型的chunk
        if 'messages' in chunk:
            print(f"\n[DEBUG] 收到消息chunk:")
            for msg in chunk['messages']:
                if hasattr(msg, 'content'):
                    print(f"[DEBUG] 消息类型: {type(msg).__name__}")
                    print(f"[DEBUG] 消息内容: {msg.content[:100]}...")  # 只显示前100字符

        # 处理 [updates] 类型的chunk
        elif 'model' in chunk:
            model_updates = chunk.get('model', {})
            if 'messages' in model_updates:
                for msg in model_updates['messages']:
                    if isinstance(msg, AIMessage) and msg.content:
                        # 这是AI的回复内容，打印出来
                        print(msg.content, end="", flush=True)
                    elif isinstance(msg, HumanMessage):
                        print(f"\n[DEBUG] 用户消息: {msg.content}")
                    else:
                        print(f"\n[DEBUG] 其他消息类型: {type(msg).__name__}")

        # 打印完整的chunk结构用于调试
        else:
            print(f"\n[DEBUG] 完整chunk结构:")
            print(json.dumps(chunk, indent=2, default=str, ensure_ascii=False)[:500] + "...")
    elif hasattr(chunk, '__dict__'):
        # 处理对象类型的chunk
        print(f"\n[DEBUG] 对象chunk: {type(chunk).__name__}")
        if hasattr(chunk, 'content'):
            print(f"[DEBUG] 内容: {chunk.content}")
            if isinstance(chunk, AIMessage):
                print(chunk.content, end="", flush=True)
    elif isinstance(chunk, str):
        # 直接字符串内容
        print(chunk, end="", flush=True)
    else:
        # 其他类型
        print(f"\n[DEBUG] 未知chunk类型: {type(chunk)}")
        print(f"[DEBUG] 值: {chunk}")

if __name__ == '__main__':

    # 运行对话
    asyncio.run(run_conversation())