# PythonProject

**Repository Path**: leocc001/python-project

## Basic Information

- **Project Name**: PythonProject
- **Description**: 学习Python，以FastAPI框架为核心，结合LangChain，LangGraph，RAG的AI应用开发基础
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: dev
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-05-02
- **Last Updated**: 2026-05-11

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

1，LangChain创建大模型的代码实现步骤
1，导包
2，创建大模型，ChatOllama()，括号里面指定model，url等等
3，准备一个提示词prompt
4，调用大模型，通过 model.invoke(prompt) 调用大模型，里面返回的值打印出来即可
2，常见组件
- Models：模型，各种类型的模型和模型集成，比如Qwen、DeepSeek
- Prompts：提示，包括提示管理、提示优化和提示序列化
- Memory：记忆，用来保存和模型交互时的上下文状态
- Indexes：索引，用来结构化文档，以便和模型交互
- Chains：链，一系列对各种组件的调用
- Agents：代理，决定模型采取哪些行动，执行并且观察流程，直到完成为止
2.1，Models组件
模型组件，消息聊天有3种类型，这三种类型都需要导包 
SystemMessage : 系统提示词
HumanMessage ：用户问题
AIMessage：AI的回答
这几个聊天封装在一个列表中
调用大模型并且使用这些组件的步骤
1，导包，导入模型包，导入消息类型的包
2，创建大模型，指定模型，url等
3，准备提示词，在准备提示词的时候，可以将这三个消息内容放入进入，以保证大模型回答的更精准无误
4，调用大模型，model.invoke(prompt)，返回的结果为大模型的输出内容

2.2，Prompts组件
在提示词的组件中，也有三种提示词的方式
字符串模板提示词 PromptTemplate
聊天场景下的提示词 ChatPrompt
引入个别例子的提示词 few-shot
2.2.1，PromptTemplate
字符串模板链，适合放入一些变量的消息
在代码中的实现步骤为
1，导包
2，创建大模型
3，提示词的准备，首先构建出字符串的模板，例如 
temple=PromptTemplate.from_template("我的邻居姓{lastname}，他生了个儿子，给他儿子起个名字")
然后的将返回的结果扔给提示词，在提示词里面填入变量的内容
prompt = template.invoke({"lastname": "王"})
4，调用大模型，model.invoke(promt)，返回的结果就是大模型的输出内容
2.2.2，ChatPrompt
聊天场景下的提示词 
在代码中的实现步骤为
1，导包
2，创建大模型
3，提示词的准备，首先构建出聊天场景提示词的模板，有系统提示词的内容，有用户问题的内容
例如 
template = ChatPromptTemplate([
    ("system", "你是一个诗人"),
    ("user", "你要写一首关于{topic}的{style}风格的诗")
])
然后的将返回的结果扔给提示词，在提示词里面填入变量的内容
prompt = template.invoke({"topic": "冬天", "style": "豪放"})
4，调用大模型，model.invoke(promt)，返回的结果就是大模型的输出内容

2.2.3，few-shot
引入少量案例的提示词方式，引导生成方式 
在代码中的实现步骤为
1，导包
2，创建大模型
3，提示词的准备，首先构建出案例的样板
例如 
examples = [
    {"word": "开心", "antonym": "难过"},
    {"word": "高", "antonym": "矮"}
]
3.1 然后准备样板模板 ，引用案例的样板中的样例
template = PromptTemplate(template="现在有一个单词是:{word},反义词:{antonym}")
3.2 在然后准备提示词模板(核心步骤)
few_shot_template = FewShotPromptTemplate(
    examples = examples,              # 提示词示例数据
    example_prompt = template,     # 提示词示例模版
    prefix="你是一个汉语言文学专家, 请给出对应单词的反义词",    
    suffix="现在有一个单词是:{w},它的反义词是:" 
)
3.3 最后在模板中填充
prompt = few_shot_template.format(w="粗")
4，调用大模型，model.invoke(promt)，返回的结果就是大模型的输出内容

2.3，Chains链组件
通过一条链进行使用 
1，导包
2，创建大模型 
3，构建提示词
3.1 构建提示词模板  
template = PromptTemplate.from_template("我的邻居姓{lastname}，他生了个儿子，给他儿子起个名字，"
3.2 将提示词模板跟大模型组装一条链
chain = template|modle|JsonOutputParser()
4，通过这条链，进行调用，传入变量值chain.invoke({"lastname": "王"}) 得到的结果就是大模型的输出 

2.4，Agents组件 
简单来说：
 👉 LLM 是“大脑”
 👉 Tool 是“手脚”
 👉 Agent 是“大脑的调度中心”
2.4.1，Tool手脚
方法上加@tool注解即可，tool中有两个默认参数 
name_or_callable（具体名称） 和description（描述）
调用的时候，直接 @tool的方法.invoke()，括号里面填写参数 
2.4.2，调用工具
以调用自定义工具为例子
1，导包
2，自定义函数@tool
3，创建大模型对象
4，定义工具，将自定义的函数，都放到一个列表中
5，构建智能体模型create_agent() 括号里面要写，模型对象，工具列表，系统提示词
6，通过 智能体.invoke() 括号里面填入参数即可 

2.5，memory组件
Memory是一个记忆模块
短期记忆
首先初始化记忆 memory=InMemorySaver()
在构建智能体模型的时候，添加短期记忆功能 checkpointer=memory
3，RAG
3.1，文档加载
加载步骤
1，导包
2，创建加载器 TextLoader，PyPDFLoader，Unstructured，MarkdownLoader，CSVLoader分别是txt，pdf，md，csv格式的，我以text为例
loader = TextLoader(file_path="../z_file/life_insurance_terms.txt", encoding="utf-8")
3，加载文档 loader.load() 得到的结果是一个个文档 
3.2，文档拆分器
# sample_text是一个文本，我这里没有写 

# 1.导包
from langchain_text_splitters import CharacterTextSplitter
# 2.创建拆分器对象
splitter = CharacterTextSplitter(
    chunk_size=100,  # 每个文本块的最大字符数
    chunk_overlap=10,  # 块之间的重叠字符数
    separator=""  # 用于拆分文本的分隔符
)
# 3.进行拆分
chunks = splitter.split_text(sample_text)
# 4.打印结果
for chunk in chunks:
    print(len(chunk))
    print(chunk)
    print("-" * 20)
3.3，向量化，向量存储，向量检索 
文档拆分后要将他变成一个个的向量 
1，创建字符拆分器对象
2，执行文本拆分
3，创建向量化模型
4，向量化并存储
# 1.创建字符拆分器对象
splitter = CharacterTextSplitter(
    separator="\n",  # 使用换行符作为分隔符
    chunk_size=150,  # 每个块最大150字符
    chunk_overlap=20,  # 块之间重叠20字符
)

# 2.执行文本拆分
chunks = splitter.split_text(sample_text)# 拆分出来是块 

# 3.创建向量化模型
embeddings_model = DashScopeEmbeddings(model="text-embedding-v4")# embeddings_model 高维数组

# 4.向量化并存储
documents = [Document(chunk) for chunk in chunks]# 获取一个个片段，转换为文档 封装成列表 
Chroma.from_documents(
    documents,                # 传入刚才封装好的文档列表
    embeddings_model,         # 传入之前创建好的模型
    persist_directory='./chroma_vsdb' # 指定本地存储路径，数据库文件会保存在这个文件夹里
    )
3.5，向量检索
步骤
1，创建向量化模型
2，连接向量库 
db = Chroma(persist_directory="./chroma_vsdb", embedding_function=embeddings_model)
3，查询
docs = db.similarity_search_with_relevance_scores(query="LangChain是什么?",k=1) 
for doc in docs:
    print(doc)

3.6，总体步骤
1，文档加载阶段
在文档加载阶段，首先判断文档是否存在，文件存在在进行判断是那种类型，是pdf还是md等等
根据类型进行加载
loader = TextLoader(file_path, encoding="utf-8")
documents = loader.load()
2，文档拆分阶段
定义一个文档拆分器的函数self.splitter调用split_documents进行拆分
chunks = self.splitter.split_documents(documents)
3，向量化存储阶段
Chroma.from_documents(
    chunks,                # 拆分结果
    self.embedding_model,  # 指定模型 
    persist_directory=self.vector_store_path # 拆分的存储路径 
)
4，检索数据库阶段
# 1. 连接数据库
db = Chroma(
    persist_directory=self.vector_store_path,    # 指定数据库路径
    embedding_function=self.embedding_model      # 指定向量化模型
)
# 2. 查询
docs = db.similarity_search_with_relevance_scores(query=query, k=config.TOP_K)
# 3. 输出
for doc in docs:
    print(doc)
4，MCP
用代码来解释MCP就是无论代码是使用Java，还是python还是C++ 都可以通过MCP协议结合起来
MCP协议有两个核心角色：客户端和服务端
MCP服务端(Tool Provider)
- 角色：工具的提供者。
- 职责：将一个或多个本地函数（@Tool）包装起来，通过个标准的MCP接口暴露出去。它监听来自客户端的请求，执行对应的函数，并返回结果。
- 例子：一个天气查询服务、一个数学计算服务、一个数据库访问服务。
MCP客户端(Tool Consumer)
- 角色：工具的调用者或消费者。
- 职责：连接到MCP服务端，查询可用的工具列表（自发现），并根据需要调用这些工具。
- 例子：大模型Agent、自动化脚本、任何需要远程执行功能的应用程序。
以SSE传输方式为例子
服务端
1，初始化 FastMCP 实例
在方法上面添加@mcp.tool注解，注解里面写工具名，工具作用
2，运行SSE mcp.run(transport="sse")
from mcp.server import FastMCP

# 初始化 FastMCP 实例
mcp = FastMCP(
    name="Weather-Server", # name: 服务器名称
    port=8001 # port: 服务器运行的端口
)

@mcp.tool(
    name="get_weather", # 工具的名
    description="获取指定城市的天气信息，入参为城市中文名称" # 工具作用
)
def get_weather(city: str) -> str:
    """获取指定城市的天气信息，入参为城市中文名称"""
    weather = {"北京": "晴朗25°C", "上海": "多云23°C", "广州": "阵雨28°C"}
    return weather.get(city, f"无{city}天气数据")


def main():
    """配置并启动 MCP SSE 服务器"""
    print("=" * 30)
    print("MCP 服务器正在启动...")
    print("当前传输协议: SSE (Server-Sent Events)")
    print("=" * 30)
    try:
        # 以 SSE 模式运行服务器
        mcp.run(transport="sse")
    except KeyboardInterrupt:
        print("\n服务器已停止")
    except Exception as e:
        print(f"服务器启动失败: {e}")


if __name__ == "__main__":
    main()

客户端（agent调用）
import asyncio
import os
import warnings
from langchain_openai import ChatOpenAI
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent


# 创建ChatOpenAI实例，配置大语言模型参数
llm = ChatOpenAI(
    model="qwen-max",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",  # 百炼的通用端点
    api_key=os.getenv("DASHSCOPE_API_KEY")
)

# 定义异步主函数
async def main():
    # 创建多服务器MCP客户端实例，配置服务器连接信息
    client = MultiServerMCPClient(
        {
            "Weather-Server": {
                "transport": "sse",  # 使用SSE传输协议
                "url": "http://127.0.0.1:8001/sse"  # 服务器SSE端点URL
            }
        }
    )
    
    # 从MCP服务器获取可用工具列表
    tools = await client.get_tools()
    
    # 打印工具信息
    print(tools)

    # 创建agent，结合LLM和工具
    agent = create_agent(
        model=llm,  # 模型名称
        tools=tools # 工具 
     )
     
    # 用户交互循环
    while True:
        # 获取用户输入
        my_input = input("请输入（quit退出交互）：")
        # 检查退出条件
        if my_input == quit":
            break
        # 异步调用代理处理用户输入
        out = await agent.ainvoke({"messages": [{"role": "user", "content": my_input}]})
        # 打印返回结果
        print(out['messages'][-1].content)


# 主程序入口点
if __name__ == "__main__":
    # 运行异步主函数
    asyncio.run(main())