python --version

virtualenv skoloenvsource skoloenv/bin/activatepip install pydantic-ai

from pydantic_ai importAgentfrom pydantic_ai.models.openai importOpenAIModel
# 引用你所需的模型model =OpenAIModel('gpt-4o', api_key='add-your-api-key-here')agent =Agent(model)result = agent.run_sync("什么是比特币？")print(result.data)

export OPENAI_API_KEY='your-api-key'

KnownModelName=Literal["openai:gpt-4o","openai:gpt-4o-mini","openai:gpt-4-turbo","openai:gpt-4","openai:o1-preview","openai:o1-mini","openai:o1","openai:gpt-3.5-turbo","groq:llama-3.3-70b-versatile","groq:llama-3.1-70b-versatile","groq:llama3-groq-70b-8192-tool-use-preview","groq:llama3-groq-8b-8192-tool-use-preview","groq:llama-3.1-70b-specdec","groq:llama-3.1-8b-instant","groq:llama-3.2-1b-preview","groq:llama-3.2-3b-preview","groq:llama-3.2-11b-vision-preview","groq:llama-3.2-90b-vision-preview","groq:llama3-70b-8192","groq:llama3-8b-8192","groq:mixtral-8x7b-32768","groq:gemma2-9b-it","groq:gemma-7b-it","gemini-1.5-flash","gemini-1.5-pro","gemini-2.0-flash-exp","vertexai:gemini-1.5-flash","vertexai:gemini-1.5-pro","mistral:mistral-small-latest","mistral:mistral-large-latest","mistral:codestral-latest","mistral:mistral-moderation-latest","ollama:codellama","ollama:gemma","ollama:gemma2","ollama:llama3","ollama:llama3.1","ollama:llama3.2","ollama:llama3.2-vision","ollama:llama3.3","ollama:mistral","ollama:mistral-nemo","ollama:mixtral","ollama:phi3","ollama:qwq","ollama:qwen","ollama:qwen2","ollama:qwen2.5","ollama:starcoder2","claude-3-5-haiku-latest","claude-3-5-sonnet-latest","claude-3-opus-latest","test",]

pip install psycopg2pip install asyncpg

import psycopg2
DB_DSN ="database-connection-string"
def create_notes_table():"""如果“notes”表不存在，则创建它，包含“id”、“title”和“text”字段。"""    create_table_query ="""    CREATE TABLE IF NOT EXISTS notes (        id SERIAL PRIMARY KEY,        title VARCHAR(200) UNIQUE NOT NULL,        text TEXT NOT NULL);"""try:        connection = psycopg2.connect(DB_DSN)        cursor = connection.cursor()        cursor.execute(create_table_query)        connection.commit()print("成功创建或验证了‘notes’表。")except psycopg2.Erroras e:print(f"创建表时出错：{e}")finally:if connection:            cursor.close()            connection.close()
def check_table_exists(table_name: str)->bool:"""检查指定表是否在数据库中存在。"""    query ="""    SELECT EXISTS (        SELECT 1        FROM information_schema.tables        WHERE table_schema ='public'        AND table_name =%s);"""try:        connection = psycopg2.connect(DB_DSN)        cursor = connection.cursor()        cursor.execute(query,(table_name,))        exists = cursor.fetchone()[0]return existsexcept psycopg2.Erroras e:print(f"检查表时出错：{e}")returnFalsefinally:if connection:            cursor.close()            connection.close()

import asyncpgfrom typing importOptional,List
classDatabaseConn:def __init__(self):"""存储用于连接的DSN（数据源名称）。"""self.dsn = DB_DSN
    async def _connect(self):"""打开与PostgreSQL的异步连接。"""return await asyncpg.connect(self.dsn)
    async def add_note(self, title: str, text: str)->bool:"""插入具有给定标题和文本的笔记。如果存在同名笔记，则不会覆盖。"""        query ="""        INSERT INTO notes (title, text)        VALUES ($1, $2)        ON CONFLICT (title) DO NOTHING;"""        conn = await self._connect()try:            result = await conn.execute(query, title, text)return result =="INSERT 0 1"finally:            await conn.close()
    async def get_note_by_title(self, title: str)->Optional[dict]:"""检索与指定标题匹配的笔记。返回字典或None。"""        query ="SELECT title, text FROM notes WHERE title = $1;"        conn = await self._connect()try:            record = await conn.fetchrow(query, title)if record:return{"title": record["title"],"text": record["text"]}returnNonefinally:            await conn.close()
    async def list_all_titles(self)->List[str]:"""获取并返回所有笔记标题。"""        query ="SELECT title FROM notes ORDER BY title;"        conn = await self._connect()try:            results = await conn.fetch(query)return[row["title"]for row in results]finally:            await conn.close()

from dataclasses import dataclassfrom pydantic importBaseModelfrom pydantic_ai importAgent,RunContextfrom typing importOptional,Listfrom database importDatabaseConnfrom pydantic_ai.models.openai importOpenAIModel
OPENAI_API_KEY ="enter-your-openai-api-key-here"
@dataclassclassNoteIntent:    action: str    title:Optional[str]=None    text:Optional[str]=None
@dataclassclassNoteDependencies:    db:DatabaseConn
classNoteResponse(BaseModel):    message: str    note:Optional[dict]=None    titles:Optional[List[str]]=None
# 1. 用于解析用户意图的代理intent_model =OpenAIModel('gpt-4o-mini', api_key=OPENAI_API_KEY)intent_agent =Agent(    intent_model,    result_type=NoteIntent,    system_prompt=("你是一个意图提取助手。理解用户想要做什么（例如创建、检索、列出），并提取相关数据，如标题和文本。""你的输出格式必须是JSON-like结构，包含键：action、title、text。"))
# 2. 用于执行识别动作的代理action_model =OpenAIModel('gpt-4o-mini', api_key=OPENAI_API_KEY)action_agent =Agent(    action_model,    deps_type=NoteDependencies,    result_type=NoteResponse,    system_prompt=("根据识别的用户意图，在笔记存储上执行请求的操作。""操作可以包括：‘create’（添加笔记）、‘retrieve’（获取笔记）或‘list’（列出所有笔记）。"))
# action_agent的工具@action_agent.toolasync def create_note_tool(ctx:RunContext[NoteDependencies], title: str, text: str)->NoteResponse:    db = ctx.deps.db    success = await db.add_note(title, text)returnNoteResponse(message="CREATED:SUCCESS"if success else"CREATED:FAILED")
@action_agent.toolasync def retrieve_note_tool(ctx:RunContext[NoteDependencies], title: str)->NoteResponse:    db = ctx.deps.db    note = await db.get_note_by_title(title)returnNoteResponse(message="GET:SUCCESS", note=note)if note elseNoteResponse(message="GET:FAILED")
@action_agent.toolasync def list_notes_tool(ctx:RunContext[NoteDependencies])->NoteResponse:    db = ctx.deps.db    all_titles = await db.list_all_titles()returnNoteResponse(message="LIST:SUCCESS", titles=all_titles)
async def handle_user_query(user_input: str, deps:NoteDependencies)->NoteResponse:# 确定用户意图    intent = await intent_agent.run(user_input)print(intent.data)if intent.data.action =="create":        query = f"创建名为‘{intent.data.title}’的笔记，文本为‘{intent.data.text}’。"        response = await action_agent.run(query, deps=deps)return response.dataelif intent.data.action =="retrieve":        query = f"检索标题为‘{intent.data.title}’的笔记。"        response = await action_agent.run(query, deps=deps)return response.dataelif intent.data.action =="list":        query ="列出所有笔记的标题。"        response = await action_agent.run(query, deps=deps)return response.dataelse:returnNoteResponse(message="无法识别的操作。")
async def ask(query: str):    db_conn =DatabaseConn()    note_deps =NoteDependencies(db=db_conn)return await handle_user_query(query, note_deps)

pip install streamlit

import asyncioimport streamlit as stfrom main import ask  # 从main.py导入ask函数
st.set_page_config(page_title="笔记管理器", layout="centered")st.title("我的笔记仪表板")st.write("在下方输入指令以创建、检索或列出笔记。")user_input = st.text_area("你想做什么？", placeholder="例如，‘创建一篇关于我周一会议的笔记。’")if st.button("提交"):ifnot user_input.strip():        st.error("请输入内容。")else:with st.spinner("正在处理..."):try:                response = asyncio.run(ask(user_input))if response.note isnotNone:                    st.success(response.message)                    st.subheader(f"笔记标题：{response.note.get('title', '')}")                    st.write(response.note.get('text','未找到内容。'))elif response.titles isnotNone:                    st.success(response.message)if response.titles:                        st.subheader("当前标题：")for t in response.titles:                            st.write(f"- {t}")else:                        st.info("尚无可用笔记。")else:                    st.info(response.message)exceptExceptionas e:                st.error(f"错误：{e}")

streamlit run app.py