fix: main.py — Создайть просто AI агент на Python с применением langchain
This commit is contained in:
@@ -1,18 +1,18 @@
|
||||
import os
|
||||
import asyncio
|
||||
from typing import List, Dict, Any
|
||||
from typing import Any, Dict, List
|
||||
|
||||
from pydantic import SecretStr
|
||||
from langchain_openai import ChatOpenAI
|
||||
from langchain_core.messages import HumanMessage, BaseMessage
|
||||
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
|
||||
from langchain.tools import tool
|
||||
|
||||
from deepagents import create_deep_agent
|
||||
from deepagents.backends import FilesystemBackend, LocalShellBackend, CompositeBackend
|
||||
from deepagents.backends import CompositeBackend, LocalShellBackend, FilesystemBackend
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Configuration
|
||||
# ----------------------------------------------------------------------
|
||||
# LLM - OpenRouter (free tier). The API key must be stored in the environment.
|
||||
# DESIGN DECISION: Use OpenRouter LLM as required by the technical constraints.
|
||||
# NECESSITY: The course forbids local LM endpoints and mandates OpenRouter for all LLM calls.
|
||||
# OPTIMALITY: Guarantees consistent API compatibility with OpenAI SDK and avoids GPU requirements.
|
||||
# ALTERNATIVES CONSIDERED: Local LM via http://localhost:1234 - rejected due to explicit prohibition.
|
||||
llm = ChatOpenAI(
|
||||
model="openai/gpt-oss-20b:free",
|
||||
base_url="https://openrouter.ai/api/v1",
|
||||
@@ -20,8 +20,7 @@ llm = ChatOpenAI(
|
||||
temperature=0.7,
|
||||
)
|
||||
|
||||
# Backend for the agents - a simple composite that allows file operations
|
||||
# and execution of shell commands inside a sandboxed workspace.
|
||||
# Backend required by deepagents - combines a shell and filesystem workspace.
|
||||
backend = CompositeBackend(
|
||||
[
|
||||
LocalShellBackend(workspace_dir="./workspace"),
|
||||
@@ -29,91 +28,80 @@ backend = CompositeBackend(
|
||||
]
|
||||
)
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Sub-agent: price generator
|
||||
# ----------------------------------------------------------------------
|
||||
def _create_price_subagent() -> Any:
|
||||
"""
|
||||
Creates a lightweight sub-agent that, given a product and a city,
|
||||
returns a markdown table with a plausible price and a store name.
|
||||
The sub-agent re-uses the same LLM and backend as the main agent.
|
||||
"""
|
||||
subagent = create_deep_agent(
|
||||
model=llm,
|
||||
tools=[], # No additional tools are required for price generation
|
||||
backend=backend,
|
||||
system_prompt=(
|
||||
"You are a price-estimation sub-agent. "
|
||||
"Given a product name and a city, generate a realistic price "
|
||||
"in Russian rubles and suggest a typical store. "
|
||||
"Return the result as a markdown table with columns: "
|
||||
"`Продукт`, `Цена (руб.)`, `Магазин`."
|
||||
),
|
||||
)
|
||||
return subagent
|
||||
|
||||
_price_subagent = _create_price_subagent()
|
||||
|
||||
@tool
|
||||
def get_price(product: str, city: str) -> str:
|
||||
"""
|
||||
Estimate the price of a product in a given city.
|
||||
The function creates a sub-agent that returns a markdown table:
|
||||
Получить примерную цену продукта в указанном городе.
|
||||
Возвращает таблицу в markdown-формате:
|
||||
| Продукт | Цена (руб.) | Магазин |
|
||||
"""
|
||||
# Build the prompt for the sub-agent
|
||||
prompt = HumanMessage(
|
||||
content=f"Продукт: {product}\nГород: {city}\nСгенерируй цену."
|
||||
# DESIGN DECISION: Sub-agent is created inside the tool using the same LLM.
|
||||
# NECESSITY: The assignment explicitly requires a hierarchical agent where a tool
|
||||
# invokes its own agent to generate realistic prices.
|
||||
# OPTIMALITY: Re-using the same LLM and backend keeps the environment consistent
|
||||
# and avoids additional dependencies.
|
||||
# ALTERNATIVES CONSIDERED: Calling an external API for prices - rejected because
|
||||
# it would break the self-contained requirement.
|
||||
sub_agent = create_deep_agent(
|
||||
model=llm,
|
||||
tools=[], # No further tools needed for price generation
|
||||
backend=backend,
|
||||
system_prompt=(
|
||||
"Ты суб-агент, который генерирует реалистичную цену продукта в заданном городе. "
|
||||
"Ответ дай в виде markdown-таблицы с колонками: Продукт, Цена (руб.), Магазин."
|
||||
),
|
||||
)
|
||||
# Invoke the sub-agent asynchronously and wait for the result
|
||||
result = asyncio.run(
|
||||
_price_subagent.ainvoke(
|
||||
{"messages": [prompt]},
|
||||
|
||||
# Формируем запрос к суб-агенту
|
||||
query = f"Сгенерируй цену для продукта '{product}' в городе '{city}'."
|
||||
# Асинхронный вызов суб-агента
|
||||
async def invoke_sub() -> Dict[str, Any]:
|
||||
return await sub_agent.ainvoke(
|
||||
{"messages": [HumanMessage(content=query)]},
|
||||
{"configurable": {"thread_id": f"price-{product}-{city}"}},
|
||||
)
|
||||
)
|
||||
# The sub-agent returns a list of messages; the last one contains the table
|
||||
final_message = result["messages"][-1]
|
||||
return final_message.content if isinstance(final_message, BaseMessage) else str(final_message)
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Main agent: shopping list planner
|
||||
# ----------------------------------------------------------------------
|
||||
main_agent = create_deep_agent(
|
||||
# Запускаем цикл событий, если уже внутри async контекста
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
result = loop.create_task(invoke_sub())
|
||||
sub_result = asyncio.run(invoke_sub())
|
||||
except RuntimeError:
|
||||
# No running loop - create one
|
||||
sub_result = asyncio.run(invoke_sub())
|
||||
|
||||
# Последнее сообщение суб-агента содержит таблицу
|
||||
price_table = sub_result["messages"][-1].content
|
||||
return price_table
|
||||
|
||||
# Главный агент
|
||||
agent = create_deep_agent(
|
||||
model=llm,
|
||||
tools=[get_price],
|
||||
backend=backend,
|
||||
system_prompt="Ты помощник по планированию покупок.",
|
||||
)
|
||||
|
||||
def format_message(msg: BaseMessage) -> str:
|
||||
"""
|
||||
Convert a LangChain message to a readable string.
|
||||
Handles normal text messages and tool calls.
|
||||
"""
|
||||
if hasattr(msg, "content") and msg.content:
|
||||
return msg.content
|
||||
# Tool call representation
|
||||
if hasattr(msg, "tool_calls") and msg.tool_calls:
|
||||
call = msg.tool_calls[0]
|
||||
name = call["name"]
|
||||
args = ", ".join(f"{k}={v!r}" for k, v in call["args"].items())
|
||||
return f"{name}({args})"
|
||||
def format_message(msg: Any) -> str:
|
||||
"""Привести сообщение к читаемому виду."""
|
||||
if isinstance(msg, AIMessage) or isinstance(msg, HumanMessage):
|
||||
return f"{msg.type.upper()}: {msg.content}"
|
||||
if isinstance(msg, ToolMessage):
|
||||
return f"TOOL CALL: {msg.name}({msg.args}) -> {msg.content}"
|
||||
# Fallback
|
||||
return str(msg)
|
||||
|
||||
async def main() -> None:
|
||||
user_query = (
|
||||
"Помоги составить список покупок: молоко, хлеб, яблоки. "
|
||||
"Я нахожусь в Казани."
|
||||
)
|
||||
result = await main_agent.ainvoke(
|
||||
user_query = "Помоги составить список покупок: молоко, хлеб, яблоки. Я нахожусь в Казани."
|
||||
result = await agent.ainvoke(
|
||||
{"messages": [HumanMessage(content=user_query)]},
|
||||
{"configurable": {"thread_id": "shopping-session-1"}},
|
||||
{"configurable": {"thread_id": "session-1"}},
|
||||
)
|
||||
# Print the whole conversation chain
|
||||
for i, msg in enumerate(result["messages"], start=1):
|
||||
print(f"--- Message {i} ---")
|
||||
print(format_message(msg))
|
||||
|
||||
# Вывод всей цепочки сообщений
|
||||
for i, message in enumerate(result["messages"]):
|
||||
print(f"--- Message {i + 1} ---")
|
||||
print(format_message(message))
|
||||
print()
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user