厦大数据库实验室博客

总结、分享、收获

实验室主页



第8章 智能体开发实战-代码和提示词-林子雨编著《AI编程》

林子雨老师
大数据学习路线图

林子雨编著《AI编程》（访问教材官网）

8.3.1 LangChain核心能力

在开始之前，先安装LangChain及其配套库：

pip install langchain==1.3.6 langchain-openai==1.3.0 python-dotenv==1.0.0

新建一个项目llm_test，在代码中，我们使用python-dotenv库来管理环境变量。dotenv可以从“.env”文件中加载环境变量，避免将敏感信息（如API密钥）硬编码到代码中。一个典型的“.env”文件配置如下：

# 大语言模型 API 配置（基于腾讯混元 / TokenHub 终端）
# -----------------------------------------------------------------
# 你的 TokenHub API 密钥
OPENAI_API_KEY="sk-uDgIrLroLUhqcH95M3o2mduso7WBL5TcYsePYRHjToRv9PS1"

# 腾讯 TokenHub 的中转 API 地址（告诉 LangChain 不要往美国 OpenAI 发请求）
OPENAI_BASE_URL="https://tokenhub.tencentmaas.com/v1"

# 默认使用的模型名称（对应你刚才 Demo 里的模型版本）
OPENAI_MODEL_NAME="deepseek-v4-pro"

下面代码展示了SystemMessage、HumanMessage、AIMessage三种消息类型的使用：

# langchain_demo2_basic.py
#!/usr/bin/env python3
"""Demo 2: LangChain 三种核心消息类型"""

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI

load_dotenv()
llm = ChatOpenAI(
    model=os.getenv("OPENAI_MODEL_NAME"),
    temperature=0.7,
    base_url=os.getenv("OPENAI_BASE_URL"),
)
agent = create_agent(
    llm, tools=[], system_prompt="你是一个友好的AI助手，请用中文简洁回答。"
)

def demo_human():
    """HumanMessage: 用户输入"""
    resp = agent.invoke({"messages": [HumanMessage(content="用一句话介绍人工智能")]})
    print(f"问：用一句话介绍人工智能\n答：{resp['messages'][-1].content}\n")

def demo_ai():
    """AIMessage: 多轮对话历史"""
    messages = [
        HumanMessage(content="中国的首都是哪里？"),
        AIMessage(content="中国的首都是北京。"),
        HumanMessage(content="那它的人口大约是多少？"),
    ]
    resp = agent.invoke({"messages": messages})
    print(f"问：那它的人口大约是多少？\n答：{resp['messages'][-1].content}\n")

def demo_system():
    """SystemMessage: 动态角色"""
    messages = [
        SystemMessage(content="你是一位诗人，用诗意语言回答。"),
        HumanMessage(content="描述春天"),
    ]
    resp = agent.invoke({"messages": messages})
    print(f"问：描述春天\n答：{resp['messages'][-1].content}\n")

if __name__ == "__main__":
    print("=== 演示 HumanMessage（用户输入） ===")
    demo_human()    
    print("=== 演示 AIMessage（多轮对话历史） ===")
    demo_ai()    
    print("=== 演示 SystemMessage（动态角色） ===")
    demo_system()

工具调用（Tool Calling）是智能体区别于普通聊天程序的核心能力。当用户的问题需要查询外部信息时，大语言模型不会直接编造答案，而是生成一个工具调用请求。开发者执行对应的Python函数，再将返回值传回大语言模型，由大语言模型根据工具结果生成最终回答。这个过程中产生了第四种消息类型——ToolMessage，它承载工具返回值重新喂给大语言模型。具体实例如下：

# langchain_demo3_tool_calling.py
#!/usr/bin/env python3
"""Demo 3: 天气助手（带思考过程展示）"""

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

load_dotenv()
llm = ChatOpenAI(
    model=os.getenv("OPENAI_MODEL_NAME"),
    temperature=0,
    base_url=os.getenv("OPENAI_BASE_URL"),
)

@tool
def get_weather(city: str) -> str:
    """查询城市天气（支持北京、上海、广州）"""
    db = {"北京": "晴，25°C", "上海": "多云，28°C", "广州": "小雨，32°C"}
    return f"{city}天气：{db.get(city, '晴转多云，26°C')}"

@tool
def get_air_quality(city: str) -> str:
    """查询城市空气质量（支持北京、上海、广州）"""
    db = {"北京": "AQI 65（良）", "上海": "AQI 42（优）", "广州": "AQI 89（良）"}
    return f"{city}空气质量：{db.get(city, 'AQI 75（良）')}"

agent = create_agent(
    llm,
    tools=[get_weather, get_air_quality],
    system_prompt="你是贴心的天气助手，帮用户查询天气和空气质量。",
    debug=True,
)

if __name__ == "__main__":
    resp = agent.invoke({"messages": [HumanMessage(content="北京天气怎么样？")]})
    print(f"最终回答：{resp['messages'][-1].content}\n")
    resp = agent.invoke({"messages": [HumanMessage(content="上海空气质量如何？")]})
    print(f"最终回答：{resp['messages'][-1].content}")

提示词模板允许开发者将可变部分（如角色、风格、问题）以占位符标记，运行时动态填充。相比字符串拼接，模板更结构化、更易维护。具体实例如下：

# langchain_demo4_prompt_templates.py
#!/usr/bin/env python3
"""Demo 4: PromptTemplate vs Message"""

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.prompts import (
    ChatPromptTemplate,
    HumanMessagePromptTemplate,
    SystemMessagePromptTemplate,
)
from langchain_openai import ChatOpenAI

load_dotenv()
llm = ChatOpenAI(
    model=os.getenv("OPENAI_MODEL_NAME"),
    temperature=0.7,
    base_url=os.getenv("OPENAI_BASE_URL"),
)
agent = create_agent(llm, tools=[])

# 方式1: 直接使用 Message
resp = agent.invoke({
    "messages": [
        SystemMessage(content="你是翻译助手，请准确翻译。"),
        HumanMessage(content="人工智能改变世界"),
    ]
})
print(f"翻译: {resp['messages'][-1].content}")

# 方式2: 使用 ChatPromptTemplate（支持动态参数）
template = ChatPromptTemplate.from_messages([
    SystemMessagePromptTemplate.from_template("你是{role}，用{style}风格回答。"),
    HumanMessagePromptTemplate.from_template("{question}"),
])
# 翻译场景
resp = agent.invoke({
    "messages": template.format_messages(
        role="翻译助手", style="准确的", question="人工智能改变世界"
    )
})
print(f"翻译: {resp['messages'][-1].content}")
# 切换角色 - 同一模板，不同角色
resp = agent.invoke({
    "messages": template.format_messages(
        role="幽默大师", style="搞笑的", question="什么是人工智能？"
    )
})
print(f"回答: {resp['messages'][-1].content}")

少样本提示（Few-Shot）通过在提示词中嵌入少量示例，让大语言模型快速理解业务逻辑。与仅靠文字描述规则不同，示例直接展示"输入→输出"的映射关系，大语言模型通过模仿示例中的推理模式来处理新问题。具体实例如下：

#langchain_demo5_few_shot.py
#!/usr/bin/env python3
"""Demo 5: Few-Shot 学习（使用 FewShotPromptTemplate）"""

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langchain_core.prompts import FewShotPromptTemplate, PromptTemplate
from langchain_openai import ChatOpenAI

load_dotenv()
llm = ChatOpenAI(
    model=os.getenv("OPENAI_MODEL_NAME"),
    temperature=0,
    base_url=os.getenv("OPENAI_BASE_URL"),
)

# 定义示例
examples = [
    {"question": "小明有5个苹果，给了小红2个，还剩几个？", "answer": "3个"},
    {"question": "书架上有10本书，借出去3本，还剩几本？", "answer": "7本"},
]

# 创建 FewShotPromptTemplate
prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=PromptTemplate.from_template("问：{question} → 答：{answer}"),
    suffix="问：{input}",
    input_variables=["input"],
)

agent = create_agent(llm, tools=[], system_prompt="请回答数学问题。")
resp = agent.invoke({
    "messages": [
        HumanMessage(
            content=prompt.format(input="篮子里有8个鸡蛋，打碎了2个，还剩几个？")
        )
    ]
})
print(f"问：篮子里有8个鸡蛋，打碎了2个，还剩几个？\n答：{resp['messages'][-1].content}")

多轮对话的核心在于"记忆"——智能体需要记住之前的对话内容。对话历史管理方式有两种：一种是手动列表管理帮助理解底层原理，另一种是使用InMemoryChatMessageHistory，它是LangChain封装好的API，使用更简洁。具体实例如下：

#langchain_demo6_chat_history.py
#!/usr/bin/env python3
"""Demo 6: 对话历史管理"""

import os
from dotenv import load_dotenv
from langchain.agents import create_agent
from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.messages import AIMessage, HumanMessage
from langchain_openai import ChatOpenAI

load_dotenv()
llm = ChatOpenAI(
    model=os.getenv("OPENAI_MODEL_NAME"),
    temperature=0.7,
    base_url=os.getenv("OPENAI_BASE_URL"),
)
agent = create_agent(llm, tools=[], system_prompt="你是友好的AI助手，请记住对话内容。")

# 方式一：手动列表管理
history = []
# 第一轮
history.append(HumanMessage(content="我叫小明，喜欢打篮球"))
resp = agent.invoke({"messages": history.copy()})
history.append(AIMessage(content=resp["messages"][-1].content))
print(f"用户：我叫小明，喜欢打篮球\nAI：{resp['messages'][-1].content}")
# 第二轮
history.append(HumanMessage(content="周末想运动，有什么建议？"))
resp = agent.invoke({"messages": history.copy()})
print(f"用户：周末想运动，有什么建议？\nAI：{resp['messages'][-1].content}")

# 方式二：InMemoryChatMessageHistory API
chat_history = InMemoryChatMessageHistory()
# 第一轮
chat_history.add_user_message("我叫小红，喜欢画画")
resp = agent.invoke({"messages": chat_history.messages})
chat_history.add_ai_message(resp["messages"][-1].content)
print(f"用户：我叫小红，喜欢画画\nAI：{resp['messages'][-1].content}")
# 第二轮（AI记住了名字和爱好）
chat_history.add_user_message("推荐一些艺术展览")
resp = agent.invoke({"messages": chat_history.messages})
print(f"用户：推荐一些艺术展览\nAI：{resp['messages'][-1].content}")

8.3.2 Streamlit界面框架

Streamlit采用"脚本重跑"模型——每次用户与界面交互，整个Python脚本从头执行。以下代码展示了Streamlit的最小结构：

# streamlit_demo1_hello.py
import streamlit as st

st.title("我的第一个 Streamlit 应用")
st.write("Hello, Streamlit!")

仅4行代码就完成了一个Web页面。st.title渲染标题，st.write渲染文字。在执行这段代码之前，要先在TRAE的Git Bash终端中执行如下命令安装Streamlit：

pip install streamlit==1.30.0

在TRAE的Git Bash终端中执行如下命令启动Streamlit：

streamlit run streamlit_demo1_hello.py

打字机效果让AI回复更加生动自然。具体实例如下：

# streamlit_demo2_typewriter.py
#!/usr/bin/env python3
"""Streamlit Demo 2: 打字机效果"""

import time
import streamlit as st

def typewriter(text, speed=0.05):
    for char in text:
        yield char
        time.sleep(speed)

st.title("打字机效果演示")
if st.button("开始打字"):
    placeholder = st.empty()
    full_text = ""
    for char in typewriter(
        "你好！\n欢迎学习 Streamlit！\n这是一个基于 Streamlit 的简单应用。"
    ):
        full_text += char
        placeholder.write(full_text)

在TRAE的Git Bash终端中执行如下命令启动Streamlit：

streamlit run streamlit_demo2_typewriter.py

Streamlit提供了st.chat_message和st.chat_input两个原生聊天组件。下面代码实现了完整的聊天界面（含模拟回复和打字机效果）：

# streamlit_demo3_mock_chat.py
#!/usr/bin/env python3
"""Streamlit Demo 3: 模拟大语言模型聊天界面（带打字机效果）"""

import time
import streamlit as st

def typewriter(text, speed=0.03):
    result = ""
    for char in text:
        result += char
        yield result
        time.sleep(speed)

def get_response(question):
    responses = {
        "你好": "你好！我是一个 AI 助手。请问有什么可以帮助你的？",
        "你是谁": "我是一个基于大语言模型的 AI 助手。",
        "什么是 Streamlit": "Streamlit 是一个用于快速构建网页应用的 Python 库。",
    }
    return responses.get(question, f"关于 '{question}'，我可以为你提供更多信息。")

st.set_page_config(page_title="AI 聊天助手", page_icon="🤖")
st.title("AI 聊天助手")

if "messages" not in st.session_state:
    st.session_state.messages = []

for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

if prompt := st.chat_input("请输入你的问题..."):
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)

    with st.chat_message("assistant"):
        response = get_response(prompt)
        placeholder = st.empty()
        for partial in typewriter(response):
            placeholder.markdown(partial)

       st.session_state.messages.append({"role": "assistant", "content": response})

现在将AI聊天助手中的模拟回复替换为真实的LangChain调用。具体代码如下：

# streamlit_demo4_langchain_chat.py
#!/usr/bin/env python3
"""Streamlit Demo 4: 连接真实大语言模型的聊天应用"""

import os
import time
import streamlit as st
from dotenv import load_dotenv
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI

def typewriter(text, speed=0.02):
    result = ""
    for char in text:
        result += char
        yield result
        time.sleep(speed)

st.set_page_config(page_title="LangChain AI 助手", page_icon="🦜")
st.title("LangChain AI 助手")

load_dotenv()

@st.cache_resource
def init_llm():
    return ChatOpenAI(
        model=os.getenv("OPENAI_MODEL_NAME"),
        temperature=0.7,
        base_url=os.getenv("OPENAI_BASE_URL"),
    )

llm = init_llm()

if "messages" not in st.session_state:
    st.session_state.messages = [
        {"role": "assistant", "content": "你好！我是你的 AI 助手。有什么我可以帮助你的吗？"}
    ]

for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

if prompt := st.chat_input("请输入你的问题..."):
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)

    with st.chat_message("assistant"):
        placeholder = st.empty()
        response = llm.invoke([HumanMessage(content=prompt)])
        full_response = response.content
        for partial in typewriter(full_response):
            placeholder.markdown(partial)

    st.session_state.messages.append({"role": "assistant", "content": full_response})

with st.sidebar:
    st.subheader("设置")
    if st.button("清除对话历史"):
        st.session_state.messages = [
            {"role": "assistant", "content": "你好！我是你的 AI 助手。有什么我可以帮助你的吗？"}
        ]
              st.rerun()

以下代码展示Plotly图表的渲染方式：

# streamlit_demo5_chart.py
import pandas as pd
import plotly.express as px
import streamlit as st

st.title("学生成绩统计")

df = pd.DataFrame({
    "分数区间": ["0-60分", "60-75分", "75-85分", "85分以上"],
    "学生人数": [8, 15, 22, 15],
})

fig = px.bar(
    df,
    x="分数区间",
    y="学生人数",
    color="分数区间",
    color_discrete_map={
        "0-60分": "#EF4444",
        "60-75分": "#F59E0B",
        "75-85分": "#8B5CF6",
        "85分以上": "#10B981",
    },
)
st.plotly_chart(fig, use_container_width=True)

8.3.3 哲学家智能体实践

在llm_test项目中，新建一个“.trae/rules/agent-requirement.md”文件，文件内容如下（可以直接从教材官网下载）：

# Agent 开发规范

## 1. 核心原则

- 所有和 Agent 的交互，不能用 `"input"` 函数，只能用 `HumanMessage`
- 所有输出必须通过 `resp['messages'][-1].content` 提取
- 保持代码简洁，优先使用函数封装单个功能

## 2. 技术栈版本（严格锁定）

langchain==1.3.6
langchain-core==1.4.3
langchain-openai==1.3.0
streamlit>=1.30.0
python-dotenv>=1.0.0


## 3. 消息类型规范

必须使用 `langchain_core.messages` 中的标准消息类型：

| 类型 | 用途 | 示例 |
|------|------|------|
| `HumanMessage` | 用户输入 | `HumanMessage(content="你好")` |
| `AIMessage` | AI 回复/对话历史 | `AIMessage(content="你好")` |
| `SystemMessage` | 系统指令/角色设定 | `SystemMessage(content="你是助手")` |
| `ToolMessage` | 工具调用结果 | `ToolMessage(content="结果", tool_call_id="xxx")` |

**禁止**：使用字符串或 `"input"` 字典直接传入 Agent。

## 4. Agent 创建规范

统一使用 `create_agent`：

```python
from langchain.agents import create_agent

agent = create_agent(
    llm,
    tools=[tool1, tool2],           # 工具列表，空列表用 tools=[]
    system_prompt="系统指令",        # 角色设定
    debug=True,                      # 调试时必须开启，观察推理过程
)

# 调用方式
resp = agent.invoke({"messages": [HumanMessage(content="用户问题")]})
answer = resp["messages"][-1].content

5. 工具定义规范

使用 `@tool` 装饰器，保持简单：

from langchain_core.tools import tool

@tool
def my_tool(param: str) -> str:
    """工具描述（必须写，Agent 靠这个理解工具用途）"""
    return f"结果: {param}"
  • 工具函数必须带类型注解
  • 返回类型统一为 `str`
  • 工具要单一职责，避免复杂逻辑

6. 提示词模板规范

需要动态参数时用 `ChatPromptTemplate`：

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.prompts import SystemMessagePromptTemplate, HumanMessagePromptTemplate

template = ChatPromptTemplate.from_messages([
    SystemMessagePromptTemplate.from_template("你是{role}，用{style}风格回答。"),
    HumanMessagePromptTemplate.from_template("{question}")
])

messages = template.format_messages(role="专家", style="严谨", question="问题")

简单场景直接用 `SystemMessage` + `HumanMessage`。

7. Few-Shot 学习规范

使用 `FewShotPromptTemplate`：

from langchain_core.prompts import FewShotPromptTemplate, PromptTemplate

examples = [
    {"input": "示例输入", "output": "示例输出"},
]

prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=PromptTemplate.from_template("输入：{input} → 输出：{output}"),
    suffix="输入：{input}",
    input_variables=["input"]
)

8. 对话历史管理规范

使用 `InMemoryChatMessageHistory`：

from langchain_core.chat_history import InMemoryChatMessageHistory

history = InMemoryChatMessageHistory()
history.add_user_message("用户的话")
history.add_ai_message("AI 的回复")
messages = history.messages  # 获取消息列表

禁止：用普通列表 `[]` 手动管理历史（demo 简化场景除外）。

9. Streamlit 集成规范

  • 对话历史用 `st.session_state` 持久化
  • LLM 初始化用 `@st.cache_resource` 避免重复创建
  • 用户输入统一走 `st.chat_input()`
  • 聊天界面用 `st.chat_message()` 组件
@st.cache_resource
def init_llm():
    return ChatOpenAI(...)

if "messages" not in st.session_state:
    st.session_state.messages = []

10. 代码风格

  • 函数封装单个功能，命名清晰
  • 避免过度抽象，三行相似代码优于一个 premature abstraction
  • 不添加不必要的错误处理、类型注解（框架边界除外）
  • 注释只写必要的关键逻辑说明 
    在TRAE的智能体交互界面的对话框中输入如下提示词并提交：

    请给我开发一个哲学家启发式对话应用

需求：

  • 用户与10位西方哲学家轮流对话（苏格拉底、柏拉图、亚里士多德、笛卡尔、康德、尼采、萨特、加缪、海德格尔、维特根斯坦）
  • 哲学家不直接回答，通过启发式提问回应，体现各自哲学思想
  • 回答≤150字，以反问结束
  • 每轮对话后轮换哲学家

界面：

  • 顶部显示当前哲学家及核心思想
  • 聊天消息展示区
  • 底部输入框
  • 侧边栏：可折叠历史记录+清除按钮

在TRAE的Git Bash终端中执行“streamlit run streamlit_demo_philosopher.py”启动应用。

8.4.1 初始化项目环境

参考6.3节的教程，执行“pre-commit install”命令配置pre-commit钩子，确保每次提交代码前都执行代码规范检查。然后，直接把6.3节中的“.pre-commit-config.yaml”文件复制到本项目根目录下，并执行如下命令把该文件也纳入Git版本管理：

git add .pre-commit-config.yaml 
git commit -a -m '添加pre-commit钩子的配置'

项目中的requirements.txt列出了项目的全部Python依赖，具体如下：

langchain==1.3.6
langchain-openai==0.2.12
streamlit==1.30.0
pandas==2.2.0
plotly==5.18.0
python-dotenv==1.0.0
pre-commit==3.6.0

在TRAE的Git Bash终端中执行以下命令安装全部依赖：

pip install -r requirements.txt

8.4.2 需求对齐与详细设计

在TRAE的智能体交互界面的对话框中输入如下提示词并提交（可以从教材官网下载提示词）：

严格遵循Superpowers七步法串行开发本项目，无确认绝不主动推进、不超前写代码、不跳过任何一个步骤。

固定流程顺序，缺一不可，仅完成当前步骤，等待我确认后再进入下一环。

1. 需求对齐时调用 /skill:brainstorming 苏格拉底追问敲定全部开发需求，输出需求文档，待确认
2. 环境隔离 /skill:using-git-worktrees 建立Git工作树/独立开发分支，确定项目目录，待确认
3. 任务拆解 /skill:writing-plans 拆分细粒度开发任务，明确文件、逻辑、校验方式，待确认
4. 逐任务开发 /skill:subagent-driven-development 单任务完成后自检复盘，不批量编写，待确认
5. TDD测试 /skill:test-driven-development 先写测试用例再实现代码，完成功能校验，待确认
6. 代码评审 /skill:requesting-code-review 排查漏洞、规范、性能问题并优化，待确认
7. 项目收尾 /skill:finishing-a-development-branch 完成整合、持久化、整理交付，流程结束

每步结束统一话术：本步骤完成，请确认进入下一步
未收到确认指令，仅限完善当前内容，禁止输出后续内容与代码
严禁自主提速、合并步骤、提前完成开发

现在开始第1步，逐条澄清本项目的需求，本项目的基本要求已经写入到AGENTS.md文件中。

8.4.8 运行项目与效果展示

以上各模块代码完成后，即可启动项目查看最终效果。在TRAE的Git Bash终端中，在项目根目录“data_pilot_agent/”下执行如下命令：

streamlit run app.py
© 2014 厦大数据库实验室