-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathdemo_integration.py
More file actions
54 lines (42 loc) · 1.64 KB
/
Copy pathdemo_integration.py
File metadata and controls
54 lines (42 loc) · 1.64 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
try:
from langchain_community.tools import DuckDuckGoSearchRun
except ImportError:
pass
import asyncio
from dotenv import load_dotenv
from core.agent import Agent
from llm.openai_provider import OpenAIProvider
from integrations.langchain_adapter import LangChainAdapter
load_dotenv()
async def demo_integrations():
print("=" * 60)
print("INTEGRATIONS DEMO (LangChain)")
print("=" * 60)
# 1. Use a LangChain tool in our Agent
try:
print("\n[LC] Loading LangChain Mock Tool...")
# Create a simple LangChain tool manually for testing
from langchain_core.tools import Tool as LCTool
def mock_search(query: str):
return "Found results for: " + query
lc_tool = LCTool(
name="mock_search",
func=mock_search,
description="A mock search tool"
)
custom_tool = LangChainAdapter.tool_from_langchain(lc_tool)
print(f" Converted to: {custom_tool.name}")
# Use MockLLM to avoid API key issues in this specific test
from demo import MockLLM
llm = MockLLM(model="mock")
agent = Agent(llm=llm, tools=[custom_tool])
print("\n[RUN] Executing search task...")
result = await agent.run("Search for 'DeepSeek API'", pattern="react")
if isinstance(result, dict) and "output" in result:
print(f"\n[RESULT] {result['output']}")
else:
print(f"\n[RESULT] {result}")
except Exception as e:
print(f"[SKIP] Integration demo skipped: {e}")
if __name__ == "__main__":
asyncio.run(demo_integrations())