AI入门实战|Agent骨架与工具调用|吃透核心概念

from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
from config import DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL

# 初始化 DeepSeek 模型
llm = ChatOpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL,
    model="deepseek-chat",
    temperature=0.1
)

# 计算器工具
@tool
def calculator(expression: str) -> str:
    """数学计算工具，接收四则运算表达式"""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"计算失败：{str(e)}"

# 会话记忆
session_store = {}
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in session_store:
        session_store[session_id] = ChatMessageHistory()
    return session_store[session_id]

# 提示词 & 执行链
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是智能助手，可以回答问题，也可以协助做数学计算。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])
chain = prompt | llm
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

# 交互式主逻辑
if __name__ == "__main__":
    print("===== 交互式对话（输入 exit 退出）=====")
    print("支持聊天、数学计算\n")
    session_id = "session_001"

    while True:
        user_text = input("你：")
        if user_text.strip().lower() == "exit":
            print("助手：再见！")
            break
        
        # 单独调用计算工具（演示）
        if any(char in user_text for char in "+-*/") and all(c not in user_text for c in "一二三四五六七八九十"):
            print(f"计算器结果：{calculator.run(user_text)}")
            continue

        # 正常对话
        res = chain_with_history.invoke(
            {"input": user_text},
            config={"session_id": session_id}
        )
        print(f"助手：{res.content}")

input("\n按回车键关闭窗口...")

from langchain_chroma import Chroma
from langchain_community.embeddings import HuggingFaceEmbeddings

# 本地向量模型
embeddings = HuggingFaceEmbeddings(model_name="all-MiniLM-L6-v2")

# 初始化向量库
vector_db = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings,
    collection_name="agent_knowledge"
)

# 写入知识库
knowledge_texts = [
    "AI Agent 核心公式：Agent = LLM + Planning + Memory + Tools",
    "Agent 运行闭环：感知 → 规划 → 行动 → 观察，循环执行直到任务结束",
    "ChromaDB 是轻量级本地向量数据库，用来实现Agent长期知识库记忆"
]
vector_db.add_texts(texts=knowledge_texts)
vector_db.persist()

# 交互式问答
if __name__ == "__main__":
    print("===== 知识库问答（输入 exit 退出）=====")
    print("可提问 AI Agent 相关问题\n")

    while True:
        user_query = input("请提问：")
        if user_query.strip().lower() == "exit":
            print("已退出问答")
            break
        
        docs = vector_db.similarity_search(user_query, k=1)
        for doc in docs:
            print(f"检索答案：{doc.page_content}\n")

input("\n按回车键关闭窗口...")

from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Sequence
import operator
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_community.chat_message_histories import ChatMessageHistory
from config import DEEPSEEK_API_KEY, DEEPSEEK_BASE_URL

# 初始化 DeepSeek
llm = ChatOpenAI(
    api_key=DEEPSEEK_API_KEY,
    base_url=DEEPSEEK_BASE_URL,
    model="deepseek-chat",
    temperature=0
)

# 全局记忆
chat_history = ChatMessageHistory()

# 计算器工具
@tool
def calculator(expression: str) -> str:
    """数学计算工具，仅处理四则运算表达式"""
    try:
        return f"计算结果：{eval(expression)}"
    except Exception:
        return "表达式错误，计算失败"

# 定义状态
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

# 规划节点（感知+判断）
def plan_node(state: AgentState) -> AgentState:
    messages = state["messages"]
    prompt_text = """
你是智能Agent，严格遵守规则：
1. 遇到数学计算，只输出：【工具:calculator|数学表达式】
2. 普通问题直接正常回答
"""
    full_messages = [HumanMessage(content=prompt_text)] + messages
    response = llm.invoke(full_messages)
    return {"messages": [response]}

# 行动节点（执行工具）
def action_node(state: AgentState) -> AgentState:
    last_msg = state["messages"][-1].content
    if "【工具:calculator|" in last_msg:
        expr = last_msg.split("|")[-1].replace("】", "").strip()
        tool_result = calculator.run(expr)
        return {"messages": [AIMessage(content=f"工具返回：{tool_result}")]}
    return {"messages": state["messages"]}

# 终止判断
def judge_end(state: AgentState) -> str:
    last_msg = state["messages"][-1].content
    if "【工具:" in last_msg:
        return "continue"
    return END

# 构建工作流
workflow = StateGraph(AgentState)
workflow.add_node("plan", plan_node)
workflow.add_node("action", action_node)
workflow.set_entry_point("plan")
workflow.add_edge("plan", "action")
workflow.add_conditional_edges(
    "action",
    judge_end,
    {"continue": "plan", END: END}
)
agent_app = workflow.compile()

# 交互式主程序
if __name__ == "__main__":
    print("===== LangGraph 智能Agent（输入 exit 退出）=====")
    print("支持聊天、自动数学计算\n")

    while True:
        user_input = input("你：")
        if user_input.strip().lower() == "exit":
            print("助手：再见！")
            break
        
        # 执行Agent流程
        final_answer = ""
        for step_output in agent_app.stream({"messages": [HumanMessage(content=user_input)]}):
            for _, state_data in step_output.items():
                final_answer = state_data["messages"][-1].content
        
        print(f"助手：{final_answer}\n")

input("\n按回车键关闭窗口...")